prop/lib/microwaveprop/workers/common_volume_radar_worker.ex

defmodule Microwaveprop.Workers.CommonVolumeRadarWorker do
  @moduledoc """
  Per-contact radar enrichment.

  For each contact, fetches the IEM n0q composite-reflectivity frame
  closest to the QSO timestamp and aggregates the pixels that fall
  inside the lens-shaped common volume between the two endpoints (the
  intersection of 400 km-radius disks). The aggregated statistics feed
  the rain-scatter vs tropo classifier.

  Jobs are unique on `contact_id` so the backfill and submission-time
  enqueue paths collapse to a single job per contact.
  """

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

  alias Microwaveprop.Propagation.CommonVolume
  alias Microwaveprop.Radio.Contact
  alias Microwaveprop.Radio.ContactCommonVolumeRadar
  alias Microwaveprop.Repo
  alias Microwaveprop.Weather.NexradClient

  require Logger

  # Radius of each station's effective rain-scatter neighborhood. Beyond
  # this, the bistatic geometry stops being plausible given typical rain
  # tops at ~15 km AGL.
  @radius_km 400.0

  # Pixel-to-dBZ thresholds (see NexradClient.pixel_to_dbz/1).
  @rain_dbz 25.0
  @heavy_rain_dbz 35.0
  @core_dbz 40.0

  # Sample every Nth pixel inside the common-volume bounding box. n0q is
  # 0.005°/px ≈ 0.5 km/px, so step=10 samples at ~5 km resolution —
  # enough to catch thunderstorm cores (typically 10+ km wide).
  @pixel_step 10

  @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) ->
        process(contact)

      contact ->
        mark_unavailable(contact)
        :ok
    end
  end

  defp process(%Contact{} = contact) do
    pos1 = to_latlon(contact.pos1)
    pos2 = to_latlon(contact.pos2)

    case CommonVolume.bounding_box(pos1, pos2, @radius_km) do
      :empty ->
        mark_unavailable(contact)
        :ok

      _bbox ->
        fetch_and_store(contact, pos1, pos2)
    end
  end

  defp fetch_and_store(%Contact{} = contact, pos1, pos2) do
    qso_at = contact.qso_timestamp
    rounded = NexradClient.round_to_5min(DateTime.from_naive!(qso_at, "Etc/UTC"))

    case NexradClient.fetch_decoded_frame(rounded) do
      {:ok, pixels, width} ->
        stats = aggregate_stats(pixels, width, pos1, pos2)
        upsert_row!(contact, rounded, stats)
        mark_status(contact, :complete)
        Logger.info("CommonVolumeRadarWorker: #{contact.id} ingested (max_dbz=#{stats.max_dbz || "nil"})")
        :ok

      {:error, reason} ->
        if permanent_error?(reason) do
          Logger.info("CommonVolumeRadarWorker: no frame for #{contact.id}: #{inspect(reason)}")
          mark_unavailable(contact)
          :ok
        else
          # Transient (5xx, timeout, connrefused) — return {:error, _} so
          # Oban retries and the failure is reflected in the job-failure
          # counters. Do NOT pin the contact :unavailable; it must remain
          # eligible for the next attempt.
          Logger.warning("CommonVolumeRadarWorker: transient error for #{contact.id}: #{inspect(reason)}")
          {:error, reason}
        end
    end
  end

  # Permanent: 4xx means the IEM archive genuinely has no frame at this
  # timestamp, so retrying is pointless. Everything else (5xx, transport
  # errors, decode failures) is treated as transient.
  defp permanent_error?("NEXRAD n0q HTTP " <> code) do
    case Integer.parse(code) do
      {status, _} when status in 400..499 -> true
      _ -> false
    end
  end

  defp permanent_error?(_), do: false

  @doc """
  Aggregate n0q pixel statistics over the common volume of the two
  endpoints. Pure function — accepts raw pixel buffer + width so it
  can be unit-tested without hitting the network.

  `step` controls the pixel sampling stride (default 10 ≈ 5 km/sample at
  the n0q resolution). Tests pass `step: 1` to hit every pixel of a
  small synthetic buffer.
  """
  @spec aggregate_stats(
          binary(),
          non_neg_integer(),
          CommonVolume.latlon(),
          CommonVolume.latlon(),
          keyword()
        ) ::
          %{
            pixel_count: non_neg_integer(),
            rain_pixel_count: non_neg_integer(),
            heavy_rain_pixel_count: non_neg_integer(),
            core_pixel_count: non_neg_integer(),
            max_dbz: float() | nil,
            mean_dbz: float() | nil,
            common_volume_km2: float(),
            coverage_pct: float() | nil
          }
  def aggregate_stats(pixels, width, pos1, pos2, opts \\ []) do
    Microwaveprop.Instrument.span([:radar, :aggregate_stats], %{}, fn ->
      do_aggregate_stats(pixels, width, pos1, pos2, opts)
    end)
  end

  defp do_aggregate_stats(pixels, width, pos1, pos2, opts) do
    step = Keyword.get(opts, :step, @pixel_step)

    case CommonVolume.bounding_box(pos1, pos2, @radius_km) do
      :empty ->
        empty_stats(CommonVolume.area_km2(pos1, pos2, @radius_km))

      %{min_lat: min_lat, max_lat: max_lat, min_lon: min_lon, max_lon: max_lon} ->
        height = div(byte_size(pixels), width)

        {x_min, y_max} = NexradClient.latlon_to_pixel(min_lat, min_lon)
        {x_max, y_min} = NexradClient.latlon_to_pixel(max_lat, max_lon)

        x_min = max(x_min, 0)
        x_max = min(x_max, width - 1)
        y_min = max(y_min, 0)
        y_max = min(y_max, height - 1)

        pixels
        |> collect_cv_pixels(
          width,
          Range.new(x_min, x_max, 1),
          Range.new(y_min, y_max, 1),
          pos1,
          pos2,
          step
        )
        |> finalize_stats(CommonVolume.area_km2(pos1, pos2, @radius_km))
    end
  end

  defp collect_cv_pixels(pixels, width, x_range, y_range, pos1, pos2, step) do
    # We accumulate three counters (in_cv, echo_count, cumulative dBZ) plus
    # max and threshold counts. Using a reduce keeps GC low on the ~10^5
    # pixel scan per contact.
    for y <- y_range.first..y_range.last//step,
        x <- x_range.first..x_range.last//step,
        reduce: init_acc() do
      acc -> visit_pixel(acc, pixels, width, x, y, pos1, pos2)
    end
  end

  defp visit_pixel(acc, pixels, width, x, y, pos1, pos2) do
    lat = 50.0 - y * 0.005
    lon = -126.0 + x * 0.005

    if CommonVolume.in_common_volume?(pos1, pos2, {lat, lon}, @radius_km) do
      offset = y * width + x
      <<_::binary-size(^offset), pixel_val::8, _::binary>> = pixels
      update_acc(acc, pixel_val)
    else
      acc
    end
  end

  defp init_acc do
    %{
      in_cv: 0,
      echo_count: 0,
      dbz_sum: 0.0,
      max_dbz: nil,
      rain: 0,
      heavy: 0,
      core: 0
    }
  end

  defp update_acc(acc, 0), do: %{acc | in_cv: acc.in_cv + 1}

  defp update_acc(acc, pixel_val) do
    dbz = NexradClient.pixel_to_dbz(pixel_val)

    %{
      acc
      | in_cv: acc.in_cv + 1,
        echo_count: acc.echo_count + 1,
        dbz_sum: acc.dbz_sum + dbz,
        max_dbz: max(acc.max_dbz || dbz, dbz),
        rain: acc.rain + if(dbz >= @rain_dbz, do: 1, else: 0),
        heavy: acc.heavy + if(dbz >= @heavy_rain_dbz, do: 1, else: 0),
        core: acc.core + if(dbz >= @core_dbz, do: 1, else: 0)
    }
  end

  defp finalize_stats(%{in_cv: 0}, cv_area) do
    empty_stats(cv_area)
  end

  defp finalize_stats(acc, cv_area) do
    mean = if acc.echo_count > 0, do: acc.dbz_sum / acc.echo_count

    %{
      pixel_count: acc.in_cv,
      rain_pixel_count: acc.rain,
      heavy_rain_pixel_count: acc.heavy,
      core_pixel_count: acc.core,
      max_dbz: acc.max_dbz,
      mean_dbz: mean,
      common_volume_km2: cv_area,
      coverage_pct: 100.0 * acc.echo_count / max(acc.in_cv, 1)
    }
  end

  defp empty_stats(cv_area) do
    %{
      pixel_count: 0,
      rain_pixel_count: 0,
      heavy_rain_pixel_count: 0,
      core_pixel_count: 0,
      max_dbz: nil,
      mean_dbz: nil,
      common_volume_km2: cv_area,
      coverage_pct: nil
    }
  end

  defp upsert_row!(%Contact{id: contact_id}, observed_at, stats) do
    %ContactCommonVolumeRadar{}
    |> ContactCommonVolumeRadar.changeset(Map.merge(stats, %{contact_id: contact_id, observed_at: observed_at}))
    |> Repo.insert(
      on_conflict: {:replace, ~w(observed_at common_volume_km2 pixel_count rain_pixel_count heavy_rain_pixel_count
            core_pixel_count max_dbz mean_dbz coverage_pct updated_at)a},
      conflict_target: :contact_id
    )
  end

  defp mark_unavailable(%Contact{} = contact), do: mark_status(contact, :unavailable)

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

  defp to_latlon(%{"lat" => lat, "lon" => lon}) when is_number(lat) and is_number(lon) do
    {lat / 1.0, lon / 1.0}
  end
end