prop/lib/microwaveprop/workers/weather_fetch_worker.ex

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

  alias Microwaveprop.Repo
  alias Microwaveprop.Weather
  alias Microwaveprop.Weather.IemClient
  alias Microwaveprop.Weather.SoundingParams
  alias Microwaveprop.Weather.Station
  alias Microwaveprop.Weather.SurfaceObservation

  require Logger

  @impl Oban.Worker
  def backoff(%Oban.Job{attempt: attempt}) do
    min(120 * Integer.pow(2, attempt - 1), _six_hours = 21_600)
  end

  @impl Oban.Worker
  def perform(%Oban.Job{args: %{"fetch_type" => "asos"} = args}) do
    %{
      "station_id" => station_id,
      "station_code" => station_code,
      "start_dt" => start_dt_str,
      "end_dt" => end_dt_str
    } = args

    {:ok, start_dt, _} = DateTime.from_iso8601(start_dt_str)
    {:ok, end_dt, _} = DateTime.from_iso8601(end_dt_str)

    case Repo.get(Station, station_id) do
      nil ->
        Logger.warning("WeatherFetch ASOS: station #{station_id} not found, skipping")
        :ok

      station ->
        if Weather.has_surface_observations?(station.id, start_dt, end_dt) do
          Logger.debug("WeatherFetch ASOS: #{station_code} already has obs for #{start_dt_str}..#{end_dt_str}")
          :ok
        else
          Logger.info("WeatherFetch ASOS: fetching #{station_code} for #{start_dt_str}..#{end_dt_str}")

          case IemClient.fetch_asos(station_code, start_dt, end_dt) do
            {:ok, rows} ->
              count = Enum.count(rows, & &1.observed_at)

              Enum.each(rows, fn row ->
                if row.observed_at, do: Weather.upsert_surface_observation(station, row)
              end)

              # Store stub so this station/window isn't retried on future backfills
              if count == 0 do
                midpoint = DateTime.add(start_dt, div(DateTime.diff(end_dt, start_dt), 2))

                %SurfaceObservation{}
                |> SurfaceObservation.changeset(%{station_id: station.id, observed_at: midpoint})
                |> Repo.insert(on_conflict: :nothing, conflict_target: [:station_id, :observed_at])
              end

              Logger.info("WeatherFetch ASOS: #{station_code} ingested #{count} observations")

              Phoenix.PubSub.broadcast(
                Microwaveprop.PubSub,
                "contact_enrichment:weather",
                {:weather_ready, %{station_id: station_id}}
              )

              :ok

            {:error, reason} ->
              Logger.warning("WeatherFetch ASOS: #{station_code} failed: #{inspect(reason)}")
              {:error, reason}
          end
        end
    end
  end

  def perform(%Oban.Job{args: %{"fetch_type" => "raob"} = args}) do
    %{
      "station_id" => station_id,
      "station_code" => station_code,
      "sounding_time" => sounding_time_str
    } = args

    {:ok, sounding_time, _} = DateTime.from_iso8601(sounding_time_str)

    case Repo.get(Station, station_id) do
      nil ->
        Logger.warning("WeatherFetch RAOB: station #{station_id} not found, skipping")
        :ok

      station ->
        if Weather.has_sounding?(station.id, sounding_time) do
          Logger.debug("WeatherFetch RAOB: #{station_code} already has sounding for #{sounding_time_str}")
          :ok
        else
          Logger.info("WeatherFetch RAOB: fetching #{station_code} for #{sounding_time_str}")

          case IemClient.fetch_raob(station_code, sounding_time) do
            {:ok, [parsed | _]} ->
              params = SoundingParams.derive(parsed.profile)

              sounding_attrs =
                if params do
                  %{
                    observed_at: parsed.observed_at,
                    profile: parsed.profile,
                    level_count: params.level_count,
                    surface_pressure_mb: params.surface_pressure_mb,
                    surface_temp_c: params.surface_temp_c,
                    surface_dewpoint_c: params.surface_dewpoint_c,
                    surface_refractivity: params.surface_refractivity,
                    min_refractivity_gradient: params.min_refractivity_gradient,
                    boundary_layer_depth_m: params.boundary_layer_depth_m,
                    precipitable_water_mm: params.precipitable_water_mm,
                    k_index: params.k_index,
                    lifted_index: params.lifted_index,
                    ducting_detected: params.ducting_detected,
                    duct_characteristics: params.duct_characteristics
                  }
                else
                  %{
                    observed_at: parsed.observed_at,
                    profile: parsed.profile,
                    level_count: length(parsed.profile)
                  }
                end

              Weather.upsert_sounding(station, sounding_attrs)

              Logger.info("WeatherFetch RAOB: #{station_code} ingested sounding with #{length(parsed.profile)} levels")

              Phoenix.PubSub.broadcast(
                Microwaveprop.PubSub,
                "contact_enrichment:weather",
                {:weather_ready, %{station_id: station_id}}
              )

              :ok

            {:ok, []} ->
              # Store stub sounding so this station/time isn't retried on future backfills
              Weather.upsert_sounding(station, %{
                observed_at: sounding_time,
                profile: [],
                level_count: 0
              })

              Logger.info("WeatherFetch RAOB: #{station_code} no data for #{sounding_time_str}, stored stub")
              :ok

            {:error, reason} ->
              Logger.warning("WeatherFetch RAOB: #{station_code} failed: #{inspect(reason)}")
              {:error, reason}
          end
        end
    end
  end
end