prop/lib/microwaveprop/workers/canadian_sounding_fetch_worker.ex

defmodule Microwaveprop.Workers.CanadianSoundingFetchWorker do
  @moduledoc """
  Twice-daily batch fetcher for Canadian radiosonde soundings via the
  University of Wyoming archive. IEM's RAOB endpoint has not been keeping
  Canadian stations current (most stalled Sep 2024), so this worker fills
  the gap — crucial because soundings drive the refractivity / ducting
  calibration in `SoundingParams.derive/1`.

  Finds every `weather_stations` row of type `sounding` whose `station_code`
  starts with `C` (Canadian ICAO prefix), drops the leading `C` to get the
  UWYO 3-letter code, fetches the latest available 00Z or 12Z sounding,
  and upserts it into `soundings`.
  """

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

  import Ecto.Query

  alias Microwaveprop.Repo
  alias Microwaveprop.Weather
  alias Microwaveprop.Weather.SoundingParams
  alias Microwaveprop.Weather.Station
  alias Microwaveprop.Weather.UwyoSoundingClient

  require Logger

  # UWYO publishes ~90 minutes after the synoptic hour
  @publish_delay_seconds 5_400

  @impl Oban.Worker
  def perform(%Oban.Job{args: args}) do
    sounding_time =
      case args do
        %{"sounding_time" => iso} when is_binary(iso) ->
          {:ok, dt, _} = DateTime.from_iso8601(iso)
          DateTime.truncate(dt, :second)

        _ ->
          most_recent_sounding_time(DateTime.utc_now())
      end

    stations = canadian_sounding_stations()

    Logger.info("CanadianSoundings: fetching #{length(stations)} stations for #{DateTime.to_iso8601(sounding_time)}")

    Enum.each(stations, &fetch_station(&1, sounding_time))

    :ok
  end

  @doc """
  Return the most recently-publishable 00Z or 12Z sounding time, given
  the current UTC instant. UWYO needs ~90 minutes after launch before the
  decoded profile lands on the site, so we step back from synoptic hour
  until we find one that's at least 90 minutes old.
  """
  @spec most_recent_sounding_time(DateTime.t()) :: DateTime.t()
  def most_recent_sounding_time(%DateTime{} = now) do
    cutoff = DateTime.add(now, -@publish_delay_seconds, :second)

    cond do
      cutoff.hour >= 12 ->
        %{cutoff | hour: 12, minute: 0, second: 0, microsecond: {0, 0}}

      cutoff.hour >= 0 and DateTime.compare(cutoff, %{cutoff | hour: 0, minute: 0, second: 0, microsecond: {0, 0}}) != :lt ->
        %{cutoff | hour: 0, minute: 0, second: 0, microsecond: {0, 0}}
    end
  end

  # ---------- Internal ----------

  defp canadian_sounding_stations do
    Repo.all(from(s in Station, where: s.station_type == "sounding" and ilike(s.station_code, "c%")))
  end

  defp fetch_station(%Station{station_code: code} = station, sounding_time) do
    uwyo_code = String.slice(code, 1..-1//1)

    if Weather.has_sounding?(station.id, sounding_time) do
      Logger.debug("CanadianSoundings: #{code} already has sounding for #{sounding_time}")
      :ok
    else
      do_fetch(station, code, uwyo_code, sounding_time)
    end
  end

  defp do_fetch(station, code, uwyo_code, sounding_time) do
    case UwyoSoundingClient.fetch_sounding(uwyo_code, sounding_time) do
      {:ok, [%{profile: profile} | _]} when profile != [] ->
        attrs = build_attrs(sounding_time, profile)
        Weather.upsert_sounding(station, attrs)

        Logger.info("CanadianSoundings: #{code} ingested sounding with #{length(profile)} levels")

      {:ok, _} ->
        Logger.info("CanadianSoundings: #{code} no data for #{sounding_time}")

      {:error, reason} ->
        Logger.warning("CanadianSoundings: #{code} failed: #{inspect(reason)}")
    end
  end

  defp build_attrs(sounding_time, profile) do
    base = %{observed_at: sounding_time, profile: profile, level_count: length(profile)}

    case SoundingParams.derive(profile) do
      nil ->
        base

      params ->
        Map.merge(base, %{
          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
        })
    end
  end
end