prop/lib/microwaveprop/workers/gefs_fetch_worker.ex

defmodule Microwaveprop.Workers.GefsFetchWorker do
  @moduledoc """
  Fetches one GEFS ensemble-mean forecast hour, extracts the CONUS
  grid, and upserts into `gefs_profiles`.

  Each `perform/1` handles a single `(run_time, forecast_hour)` pair so
  a failure at one hour doesn't block the others — the cron seeder
  enqueues the full `f024..f168` set per run. Per-hour independence
  also keeps memory bounded: the wgrib2 extraction step for a GEFS
  0.5° file is small (<1 MB after byte-range slicing) but the target
  0.125° grid still carries ~200k cells.
  """

  use Oban.Worker,
    queue: :gefs,
    max_attempts: 5,
    unique: [period: 3600, states: [:available, :scheduled, :executing, :retryable]]

  alias Microwaveprop.Weather
  alias Microwaveprop.Weather.GefsClient
  alias Microwaveprop.Weather.SoundingParams

  require Logger

  @valid_run_hours [0, 6, 12, 18]

  @impl Oban.Worker
  def perform(%Oban.Job{args: %{"run_time" => run_time_iso, "forecast_hour" => fh}}) when is_integer(fh) do
    with {:ok, run_time, _} <- DateTime.from_iso8601(run_time_iso),
         :ok <- validate_run_hour(run_time.hour) do
      run_date = DateTime.to_date(run_time)
      run_hour = run_time.hour

      Logger.info("GefsFetch: run_time=#{run_time_iso} fh=#{fh}")

      case GefsClient.fetch_grid_profiles(run_date, run_hour, fh) do
        {:ok, profiles} ->
          attrs = Enum.map(profiles, &build_profile_attrs(run_time, fh, &1))
          {count, _} = Weather.upsert_gefs_profiles_batch(attrs)
          Logger.info("GefsFetch: stored #{count}/#{length(attrs)} profiles for fh=#{fh}")
          :ok

        {:error, :wgrib2_not_available} ->
          Logger.warning("GefsFetch: wgrib2 missing on host — cancelling job")
          {:cancel, :wgrib2_not_available}

        {:error, reason} ->
          handle_error(reason, "fh=#{fh} @ #{run_time_iso}")
      end
    else
      _ -> {:cancel, :invalid_args}
    end
  end

  def perform(%Oban.Job{args: _args}), do: {:cancel, :invalid_args}

  @doc """
  Converts a `GefsClient.build_profile/1` result (already tagged with
  `lat` and `lon`) into a full `gefs_profiles` row attr map. Exposed
  for unit testing.
  """
  @spec build_profile_attrs(DateTime.t(), non_neg_integer(), map()) :: map()
  def build_profile_attrs(%DateTime{} = run_time, forecast_hour, profile) when is_integer(forecast_hour) do
    valid_time = DateTime.add(run_time, forecast_hour * 3600, :second)
    derived = SoundingParams.derive(profile.profile || [])

    base = %{
      run_time: DateTime.truncate(run_time, :second),
      forecast_hour: forecast_hour,
      valid_time: DateTime.truncate(valid_time, :second),
      lat: profile.lat,
      lon: profile.lon,
      profile: profile.profile,
      surface_temp_c: profile.surface_temp_c,
      surface_dewpoint_c: profile.surface_dewpoint_c,
      surface_pressure_mb: profile.surface_pressure_mb,
      pwat_mm: profile[:pwat_mm],
      wind_u_mps: profile[:wind_u_mps],
      wind_v_mps: profile[:wind_v_mps],
      cloud_cover_pct: profile[:cloud_cover_pct],
      precip_mm: profile[:precip_mm]
    }

    case derived do
      nil ->
        base

      params ->
        Map.merge(base, %{
          surface_refractivity: params.surface_refractivity,
          min_refractivity_gradient: params.min_refractivity_gradient,
          ducting_detected: params.ducting_detected,
          duct_characteristics: params.duct_characteristics
        })
    end
  end

  defp validate_run_hour(hour) when hour in @valid_run_hours, do: :ok
  defp validate_run_hour(_), do: {:error, :invalid_run_hour}

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

  defp transient?(%{__exception__: true}), do: true
  defp transient?("GEFS idx HTTP " <> status), do: server_error?(status)
  defp transient?("GEFS grib HTTP " <> status), do: server_error?(status)
  defp transient?(_), 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
end