prop/lib/microwaveprop/workers/mechanism_classify_worker.ex

defmodule Microwaveprop.Workers.MechanismClassifyWorker do
  @moduledoc """
  Per-contact propagation-mechanism classification.

  Assembles the full input map for `MechanismClassifier`:

    * `user_declared_prop_mode` — from the contact's ADIF PROP_MODE
    * `duct_either_endpoint` — from the HRRR profile at either endpoint
    * `native_best_duct_ghz` — from the nearest `hrrr_native_profiles` row
    * `radar` — from the pre-computed `contact_common_volume_radar` row
    * `kp_index` — from the daily `solar_indices` row
    * `foes_mhz` — from the nearest `ionosonde_observations` row
    * `active_meteor_shower` — looked up from the static shower calendar

  Results land on the contact as `propagation_mechanism` (string),
  `propagation_mechanism_confidence` (enum), and
  `propagation_mechanism_status`.

  Unique on `contact_id` so the submit-time enqueue path and the
  backfill cron collapse to a single job per contact.
  """

  use Oban.Worker,
    queue: :mechanism,
    max_attempts: 3,
    unique: [
      period: :infinity,
      states: [:available, :scheduled, :executing, :retryable],
      keys: [:contact_id]
    ]

  import Ecto.Query

  alias Microwaveprop.Ionosphere.Observation, as: IonosondeObservation
  alias Microwaveprop.Propagation.MechanismClassifier
  alias Microwaveprop.Radio.Contact
  alias Microwaveprop.Radio.ContactCommonVolumeRadar
  alias Microwaveprop.Repo
  alias Microwaveprop.Weather.HrrrProfile

  require Logger

  @impl Oban.Worker
  def perform(%Oban.Job{args: %{"contact_id" => contact_id}}) do
    case Repo.get(Contact, contact_id) do
      nil ->
        :ok

      %Contact{pos1: p1, pos2: p2} = contact when is_map(p1) and is_map(p2) ->
        classify_and_persist(contact)

      contact ->
        mark_status(contact, :unavailable, nil, nil)
        :ok
    end
  end

  defp classify_and_persist(%Contact{} = contact) do
    inputs = build_inputs(contact)
    %{mechanism: mechanism, confidence: confidence} = MechanismClassifier.classify(inputs)

    mark_status(contact, :complete, Atom.to_string(mechanism), confidence)
    :ok
  rescue
    e ->
      Logger.warning("MechanismClassifyWorker: failed for contact #{contact.id}: #{inspect(e)}")

      mark_status(contact, :failed, nil, nil)
      :ok
  end

  defp build_inputs(%Contact{} = contact) do
    %{
      band_mhz: Decimal.to_integer(contact.band),
      distance_km: distance_km(contact),
      qso_timestamp: contact.qso_timestamp,
      pos1: contact.pos1,
      pos2: contact.pos2,
      user_declared_prop_mode: contact.user_declared_prop_mode,
      radar: radar_for(contact),
      duct_either_endpoint: duct_at_either_endpoint?(contact),
      native_best_duct_ghz: native_duct_at_either_endpoint(contact),
      kp_index: kp_for(contact),
      foes_mhz: foes_for(contact),
      active_meteor_shower: active_shower_at(contact.qso_timestamp)
    }
  end

  defp distance_km(%Contact{distance_km: nil}), do: 0.0
  defp distance_km(%Contact{distance_km: d}), do: Decimal.to_float(d)

  defp radar_for(%Contact{id: id}) do
    case Repo.get_by(ContactCommonVolumeRadar, contact_id: id) do
      nil ->
        nil

      row ->
        %{
          max_dbz: row.max_dbz,
          heavy_rain_pixel_count: row.heavy_rain_pixel_count,
          coverage_pct: row.coverage_pct
        }
    end
  end

  defp duct_at_either_endpoint?(%Contact{pos1: p1, pos2: p2, qso_timestamp: ts}) do
    Enum.any?([p1, p2], fn pos ->
      case nearest_hrrr(pos, ts) do
        nil -> false
        %HrrrProfile{ducting_detected: true} -> true
        _ -> false
      end
    end)
  end

  defp native_duct_at_either_endpoint(%Contact{pos1: p1, pos2: p2, qso_timestamp: ts}) do
    [p1, p2]
    |> Enum.map(&nearest_native_duct(&1, ts))
    |> Enum.reject(&is_nil/1)
    |> case do
      [] -> nil
      values -> Enum.max(values)
    end
  end

  defp nearest_hrrr(pos, ts) do
    lat = pos["lat"]
    lon = pos["lon"]

    if lat && lon do
      Repo.one(
        from(h in HrrrProfile,
          where:
            h.lat >= ^(lat - 0.07) and h.lat <= ^(lat + 0.07) and h.lon >= ^(lon - 0.07) and h.lon <= ^(lon + 0.07) and
              h.valid_time >= ^DateTime.add(ts, -3600, :second) and h.valid_time <= ^DateTime.add(ts, 3600, :second),
          order_by: fragment("ABS(EXTRACT(EPOCH FROM ? - ?))", h.valid_time, ^ts),
          limit: 1
        )
      )
    end
  end

  defp nearest_native_duct(pos, ts) do
    lat = pos["lat"]
    lon = pos["lon"]

    if lat && lon do
      Repo.one(
        from(h in "hrrr_native_profiles",
          where:
            h.lat >= ^(lat - 0.07) and h.lat <= ^(lat + 0.07) and h.lon >= ^(lon - 0.07) and h.lon <= ^(lon + 0.07) and
              h.valid_time >= ^DateTime.add(ts, -3600, :second) and h.valid_time <= ^DateTime.add(ts, 3600, :second),
          order_by: fragment("ABS(EXTRACT(EPOCH FROM ? - ?))", h.valid_time, ^ts),
          limit: 1,
          select: h.best_duct_band_ghz
        )
      )
    end
  end

  defp kp_for(%Contact{qso_timestamp: ts}) do
    date = DateTime.to_date(ts)

    # solar_indices.kp_values is the daily array of 3-hour Kp readings.
    # Take the day's peak — that's the signal that matters for aurora.
    from(s in "solar_indices", where: s.date == ^date, select: s.kp_values, limit: 1)
    |> Repo.one()
    |> case do
      nil -> nil
      [] -> nil
      values when is_list(values) -> values |> Enum.reject(&is_nil/1) |> peak_or_nil()
    end
  end

  defp peak_or_nil([]), do: nil
  defp peak_or_nil(values), do: values |> Enum.max() |> round()

  # Nearest ionosonde observation within ±1h of QSO timestamp. We don't
  # restrict by location — foEs is spatially noisy but at least the
  # "nearest station in the Americas" will catch US-continental Es events.
  defp foes_for(%Contact{qso_timestamp: ts}) do
    Repo.one(
      from(io in IonosondeObservation,
        where: io.valid_time >= ^DateTime.add(ts, -3600, :second) and io.valid_time <= ^DateTime.add(ts, 3600, :second),
        where: not is_nil(io.fo_es_mhz),
        order_by: fragment("ABS(EXTRACT(EPOCH FROM ? - ?))", io.valid_time, ^ts),
        limit: 1,
        select: io.fo_es_mhz
      )
    )
  end

  # Static meteor-shower calendar (peak dates, approximate). Any contact
  # within ±3 days of a peak counts as "during the shower."
  @showers [
    {~D[2024-01-04], "Quadrantids"},
    {~D[2024-04-22], "Lyrids"},
    {~D[2024-05-06], "Eta Aquariids"},
    {~D[2024-07-30], "Southern Delta Aquariids"},
    {~D[2024-08-12], "Perseids"},
    {~D[2024-10-21], "Orionids"},
    {~D[2024-11-17], "Leonids"},
    {~D[2024-12-14], "Geminids"},
    {~D[2024-12-22], "Ursids"}
  ]

  defp active_shower_at(%DateTime{} = ts) do
    date = DateTime.to_date(ts)

    Enum.find_value(@showers, fn {peak_mmdd, name} ->
      diff = abs(Date.diff(peak_mmdd, %{date | year: peak_mmdd.year}))
      if diff <= 3, do: name
    end)
  end

  defp mark_status(%Contact{} = contact, status, mechanism, confidence) do
    contact
    |> Ecto.Changeset.change(%{
      propagation_mechanism_status: status,
      propagation_mechanism: mechanism,
      propagation_mechanism_confidence: confidence
    })
    |> Repo.update!()
  end
end