Add QsoWeatherEnqueueWorker cron job to auto-fetch weather for new QSOs

Oban cron worker runs every 4 hours, finds QSOs without weather data,
discovers nearby ASOS/sounding stations, and enqueues WeatherFetchWorker
jobs. Migrations run automatically on app start in production.

config/config.exs

@@ -44,12 +44,13 @@ config :microwaveprop, MicrowavepropWeb.Endpoint,
 
 config :microwaveprop, Oban,
   repo: Microwaveprop.Repo,
-  queues: [solar: 1, weather: 3],
+  queues: [solar: 1, weather: 3, enqueue: 1],
   plugins: [
     {Oban.Plugins.Pruner, max_age: 3600 * 24},
     {Oban.Plugins.Cron,
      crontab: [
-       {"0 8 * * *", Microwaveprop.Workers.SolarIndexWorker}
+       {"0 8 * * *", Microwaveprop.Workers.SolarIndexWorker},
+       {"0 */4 * * *", Microwaveprop.Workers.QsoWeatherEnqueueWorker}
      ]}
   ]
 

lib/microwaveprop/application.ex

@@ -7,6 +7,8 @@ defmodule Microwaveprop.Application do
 
   @impl true
   def start(_type, _args) do
+    migrate()
+
     children = [
       MicrowavepropWeb.Telemetry,
       Microwaveprop.Repo,
@@ -25,6 +27,10 @@ defmodule Microwaveprop.Application do
 
   # Tell Phoenix to update the endpoint configuration
   # whenever the application is updated.
+  defp migrate do
+    Microwaveprop.Release.migrate()
+  end
+
   @impl true
   def config_change(changed, _new, removed) do
     MicrowavepropWeb.Endpoint.config_change(changed, removed)

lib/microwaveprop/radio.ex

@@ -43,6 +43,20 @@ defmodule Microwaveprop.Radio do
     {field, dir}
   end
 
+  def unprocessed_qsos(limit \\ 500) do
+    Qso
+    |> where([q], q.weather_queued == false and not is_nil(q.pos1))
+    |> order_by([q], asc: q.qso_timestamp)
+    |> limit(^limit)
+    |> Repo.all()
+  end
+
+  def mark_weather_queued!(qso_ids) do
+    Qso
+    |> where([q], q.id in ^qso_ids)
+    |> Repo.update_all(set: [weather_queued: true])
+  end
+
   def get_qso!(id) do
     Repo.get!(Qso, id)
   end

lib/microwaveprop/radio/qso.ex

@@ -18,6 +18,7 @@ defmodule Microwaveprop.Radio.Qso do
     field :mode, :string
     field :band, :decimal
     field :distance_km, :decimal
+    field :weather_queued, :boolean, default: false
 
     timestamps(type: :utc_datetime)
   end

lib/microwaveprop/weather.ex

@@ -91,6 +91,39 @@ defmodule Microwaveprop.Weather do
     |> MapSet.new()
   end
 
+  def nearby_stations(lat, lon, station_type, radius_km) do
+    dlat = radius_km / @km_per_deg_lat
+    dlon = radius_km / (@km_per_deg_lat * :math.cos(lat * :math.pi() / 180))
+
+    Station
+    |> where([s], s.station_type == ^station_type)
+    |> where(
+      [s],
+      s.lat >= ^(lat - dlat) and s.lat <= ^(lat + dlat) and
+        s.lon >= ^(lon - dlon) and s.lon <= ^(lon + dlon)
+    )
+    |> Repo.all()
+  end
+
+  def sounding_times_around(dt) do
+    date = DateTime.to_date(dt)
+
+    times =
+      if dt.hour < 12 do
+        [
+          DateTime.new!(Date.add(date, -1), ~T[12:00:00], "Etc/UTC"),
+          DateTime.new!(date, ~T[00:00:00], "Etc/UTC")
+        ]
+      else
+        [
+          DateTime.new!(date, ~T[00:00:00], "Etc/UTC"),
+          DateTime.new!(date, ~T[12:00:00], "Etc/UTC")
+        ]
+      end
+
+    Enum.uniq(times)
+  end
+
   def weather_for_qso(qso_params, opts \\ []) do
     lat = qso_params[:lat] || qso_params.lat
     lon = qso_params[:lon] || qso_params.lon

lib/microwaveprop/workers/qso_weather_enqueue_worker.ex

@@ -0,0 +1,77 @@
+defmodule Microwaveprop.Workers.QsoWeatherEnqueueWorker do
+  @moduledoc false
+  use Oban.Worker, queue: :enqueue, max_attempts: 3
+
+  alias Microwaveprop.Radio
+  alias Microwaveprop.Weather
+  alias Microwaveprop.Workers.WeatherFetchWorker
+
+  @asos_radius_km 150
+  @sounding_radius_km 300
+
+  @impl Oban.Worker
+  def perform(%Oban.Job{}) do
+    qsos = Radio.unprocessed_qsos()
+
+    if qsos != [] do
+      jobs = build_weather_jobs(qsos)
+
+      if jobs != [] do
+        Oban.insert_all(jobs)
+      end
+
+      qso_ids = Enum.map(qsos, & &1.id)
+      Radio.mark_weather_queued!(qso_ids)
+    end
+
+    :ok
+  end
+
+  def build_weather_jobs(qsos) do
+    qsos
+    |> Enum.flat_map(&jobs_for_qso/1)
+    |> Enum.uniq_by(fn changeset -> changeset.changes.args end)
+  end
+
+  defp jobs_for_qso(qso) do
+    lat = qso.pos1["lat"]
+    lon = qso.pos1["lon"] || qso.pos1["lng"]
+
+    asos_jobs = build_asos_jobs(lat, lon, qso.qso_timestamp)
+    raob_jobs = build_raob_jobs(lat, lon, qso.qso_timestamp)
+
+    asos_jobs ++ raob_jobs
+  end
+
+  defp build_asos_jobs(lat, lon, timestamp) do
+    start_dt = DateTime.add(timestamp, -2 * 3600, :second)
+    end_dt = DateTime.add(timestamp, 2 * 3600, :second)
+
+    lat
+    |> Weather.nearby_stations(lon, "asos", @asos_radius_km)
+    |> Enum.map(fn station ->
+      WeatherFetchWorker.new(%{
+        "fetch_type" => "asos",
+        "station_id" => station.id,
+        "station_code" => station.station_code,
+        "start_dt" => DateTime.to_iso8601(start_dt),
+        "end_dt" => DateTime.to_iso8601(end_dt)
+      })
+    end)
+  end
+
+  defp build_raob_jobs(lat, lon, timestamp) do
+    sounding_times = Weather.sounding_times_around(timestamp)
+
+    stations = Weather.nearby_stations(lat, lon, "sounding", @sounding_radius_km)
+
+    for station <- stations, sounding_time <- sounding_times do
+      WeatherFetchWorker.new(%{
+        "fetch_type" => "raob",
+        "station_id" => station.id,
+        "station_code" => station.station_code,
+        "sounding_time" => DateTime.to_iso8601(sounding_time)
+      })
+    end
+  end
+end

priv/repo/migrations/20260329190247_add_weather_queued_to_qsos.exs

@@ -0,0 +1,15 @@
+defmodule Microwaveprop.Repo.Migrations.AddWeatherQueuedToQsos do
+  use Ecto.Migration
+
+  def change do
+    alter table(:qsos) do
+      add :weather_queued, :boolean, default: false, null: false
+    end
+
+    # Backfill existing rows — they were already processed by the import script
+    execute "UPDATE qsos SET weather_queued = true", "SELECT 1"
+
+    # Partial index for fast lookup of unprocessed QSOs
+    create index(:qsos, [:weather_queued], where: "weather_queued = false")
+  end
+end

test/microwaveprop/radio_test.exs

@@ -100,6 +100,68 @@ defmodule Microwaveprop.RadioTest do
     end
   end
 
+  describe "unprocessed_qsos/1" do
+    test "returns QSOs where weather_queued is false and pos1 is not nil" do
+      q1 = create_qso(%{station1: "W5XD", qso_timestamp: ~U[2026-03-28 12:00:00Z]})
+      _q2 = create_qso(%{station1: "K5TR", pos1: nil, qso_timestamp: ~U[2026-03-28 13:00:00Z]})
+
+      results = Radio.unprocessed_qsos()
+      ids = Enum.map(results, & &1.id)
+      assert q1.id in ids
+    end
+
+    test "excludes QSOs already marked as weather_queued" do
+      q = create_qso()
+      Radio.mark_weather_queued!([q.id])
+
+      assert Radio.unprocessed_qsos() == []
+    end
+
+    test "orders by qso_timestamp ascending" do
+      q_late = create_qso(%{station1: "LATE", qso_timestamp: ~U[2026-03-28 20:00:00Z]})
+      q_early = create_qso(%{station1: "EARLY", qso_timestamp: ~U[2026-03-28 10:00:00Z]})
+
+      results = Radio.unprocessed_qsos()
+      ids = Enum.map(results, & &1.id)
+      assert ids == [q_early.id, q_late.id]
+    end
+
+    test "respects limit parameter" do
+      for i <- 1..5 do
+        ts = DateTime.add(~U[2026-01-01 00:00:00Z], i * 3600, :second)
+        create_qso(%{qso_timestamp: ts})
+      end
+
+      assert length(Radio.unprocessed_qsos(3)) == 3
+    end
+  end
+
+  describe "mark_weather_queued!/1" do
+    test "sets weather_queued to true for given IDs" do
+      q1 = create_qso(%{station1: "A1A"})
+      q2 = create_qso(%{station1: "B2B"})
+
+      Radio.mark_weather_queued!([q1.id, q2.id])
+
+      assert Repo.get!(Qso, q1.id).weather_queued == true
+      assert Repo.get!(Qso, q2.id).weather_queued == true
+    end
+
+    test "does not affect other QSOs" do
+      q1 = create_qso(%{station1: "A1A"})
+      q2 = create_qso(%{station1: "B2B"})
+
+      Radio.mark_weather_queued!([q1.id])
+
+      assert Repo.get!(Qso, q1.id).weather_queued == true
+      assert Repo.get!(Qso, q2.id).weather_queued == false
+    end
+
+    test "handles empty list" do
+      assert Radio.mark_weather_queued!([]) == {0, nil}
+    end
+  end
+
   describe "get_qso!/1" do
     test "returns a QSO by ID" do
       qso = create_qso()

test/microwaveprop/weather_test.exs

@@ -278,6 +278,97 @@ defmodule Microwaveprop.WeatherTest do
     end
   end
 
+  describe "nearby_stations/4" do
+    test "returns stations within radius" do
+      {:ok, _near} =
+        Weather.find_or_create_station(%{
+          station_code: "KDFW",
+          station_type: "asos",
+          name: "DFW Airport",
+          lat: 32.90,
+          lon: -97.04
+        })
+
+      {:ok, _far} =
+        Weather.find_or_create_station(%{
+          station_code: "KORD",
+          station_type: "asos",
+          name: "Chicago O'Hare",
+          lat: 41.98,
+          lon: -87.90
+        })
+
+      stations = Weather.nearby_stations(32.85, -97.05, "asos", 50)
+      codes = Enum.map(stations, & &1.station_code)
+      assert "KDFW" in codes
+      refute "KORD" in codes
+    end
+
+    test "filters by station type" do
+      {:ok, _asos} =
+        Weather.find_or_create_station(%{
+          station_code: "KDFW",
+          station_type: "asos",
+          name: "DFW Airport",
+          lat: 32.90,
+          lon: -97.04
+        })
+
+      {:ok, _sounding} =
+        Weather.find_or_create_station(%{
+          station_code: "FWD",
+          station_type: "sounding",
+          name: "Fort Worth",
+          lat: 32.83,
+          lon: -97.30
+        })
+
+      stations = Weather.nearby_stations(32.85, -97.10, "sounding", 100)
+      codes = Enum.map(stations, & &1.station_code)
+      assert "FWD" in codes
+      refute "KDFW" in codes
+    end
+
+    test "returns empty list when no stations in radius" do
+      {:ok, _} =
+        Weather.find_or_create_station(%{
+          station_code: "KDFW",
+          station_type: "asos",
+          name: "DFW Airport",
+          lat: 32.90,
+          lon: -97.04
+        })
+
+      assert Weather.nearby_stations(45.0, -80.0, "asos", 50) == []
+    end
+  end
+
+  describe "sounding_times_around/1" do
+    test "returns bracketing 00Z and 12Z for afternoon UTC" do
+      dt = ~U[2026-03-28 18:00:00Z]
+      times = Weather.sounding_times_around(dt)
+      assert times == [~U[2026-03-28 00:00:00Z], ~U[2026-03-28 12:00:00Z]]
+    end
+
+    test "returns bracketing times for morning UTC" do
+      dt = ~U[2026-03-28 06:00:00Z]
+      times = Weather.sounding_times_around(dt)
+      assert times == [~U[2026-03-27 12:00:00Z], ~U[2026-03-28 00:00:00Z]]
+    end
+
+    test "returns single time at exactly 00Z" do
+      dt = ~U[2026-03-28 00:00:00Z]
+      times = Weather.sounding_times_around(dt)
+      assert times == [~U[2026-03-27 12:00:00Z], ~U[2026-03-28 00:00:00Z]]
+    end
+
+    test "returns single time at exactly 12Z" do
+      dt = ~U[2026-03-28 12:00:00Z]
+      times = Weather.sounding_times_around(dt)
+      assert times == [~U[2026-03-28 00:00:00Z], ~U[2026-03-28 12:00:00Z]]
+    end
+  end
+
   describe "get_solar_index/1" do
     test "returns the solar index record for a given date" do
       {:ok, record} = Weather.upsert_solar_index(%{date: ~D[2024-06-15], sfi: 150.2, sunspot_number: 120})

test/microwaveprop/workers/qso_weather_enqueue_worker_test.exs

@@ -0,0 +1,161 @@
+defmodule Microwaveprop.Workers.QsoWeatherEnqueueWorkerTest do
+  use Microwaveprop.DataCase, async: true
+
+  alias Microwaveprop.Radio.Qso
+  alias Microwaveprop.Weather
+  alias Microwaveprop.Weather.IemClient
+  alias Microwaveprop.Workers.QsoWeatherEnqueueWorker
+
+  @qso_attrs %{
+    station1: "W5XD",
+    station2: "K5TR",
+    qso_timestamp: ~U[2026-03-28 18:00:00Z],
+    mode: "CW",
+    band: Decimal.new("1296"),
+    grid1: "EM12",
+    grid2: "EM00",
+    pos1: %{"lat" => 32.9, "lon" => -97.0},
+    pos2: %{"lat" => 30.3, "lon" => -97.7},
+    distance_km: Decimal.new("295")
+  }
+
+  defp create_qso(attrs \\ %{}) do
+    {:ok, qso} =
+      %Qso{}
+      |> Qso.changeset(Map.merge(@qso_attrs, attrs))
+      |> Repo.insert()
+
+    qso
+  end
+
+  defp create_asos_station(code, lat, lon) do
+    {:ok, station} =
+      Weather.find_or_create_station(%{
+        station_code: code,
+        station_type: "asos",
+        name: "#{code} Station",
+        lat: lat,
+        lon: lon
+      })
+
+    station
+  end
+
+  defp create_sounding_station(code, lat, lon) do
+    {:ok, station} =
+      Weather.find_or_create_station(%{
+        station_code: code,
+        station_type: "sounding",
+        name: "#{code} Sounding",
+        lat: lat,
+        lon: lon
+      })
+
+    station
+  end
+
+  describe "build_weather_jobs/1" do
+    test "builds ASOS jobs for nearby stations" do
+      station = create_asos_station("KDFW", 32.90, -97.04)
+      qso = create_qso()
+
+      jobs = QsoWeatherEnqueueWorker.build_weather_jobs([qso])
+
+      asos_jobs =
+        Enum.filter(jobs, fn j ->
+          j.changes.args["fetch_type"] == "asos"
+        end)
+
+      assert length(asos_jobs) >= 1
+
+      job = hd(asos_jobs)
+      assert job.changes.args["station_id"] == station.id
+      assert job.changes.args["station_code"] == "KDFW"
+      assert job.changes.args["start_dt"]
+      assert job.changes.args["end_dt"]
+    end
+
+    test "builds RAOB jobs for nearby sounding stations" do
+      station = create_sounding_station("FWD", 32.83, -97.30)
+      qso = create_qso()
+
+      jobs = QsoWeatherEnqueueWorker.build_weather_jobs([qso])
+
+      raob_jobs =
+        Enum.filter(jobs, fn j ->
+          j.changes.args["fetch_type"] == "raob"
+        end)
+
+      assert length(raob_jobs) >= 1
+
+      job = hd(raob_jobs)
+      assert job.changes.args["station_id"] == station.id
+      assert job.changes.args["station_code"] == "FWD"
+      assert job.changes.args["sounding_time"]
+    end
+
+    test "deduplicates jobs with identical args" do
+      _station = create_asos_station("KDFW", 32.90, -97.04)
+
+      # Two QSOs at the exact same time produce identical args → deduplicated
+      q1 = create_qso(%{station1: "A1", qso_timestamp: ~U[2026-03-28 18:00:00Z]})
+      q2 = create_qso(%{station1: "A2", qso_timestamp: ~U[2026-03-28 18:00:00Z]})
+
+      jobs = QsoWeatherEnqueueWorker.build_weather_jobs([q1, q2])
+
+      asos_jobs =
+        Enum.filter(jobs, fn j ->
+          j.changes.args["fetch_type"] == "asos"
+        end)
+
+      assert length(asos_jobs) == 1
+    end
+
+    test "returns empty list when no stations nearby" do
+      qso = create_qso(%{pos1: %{"lat" => 60.0, "lon" => -150.0}})
+
+      assert QsoWeatherEnqueueWorker.build_weather_jobs([qso]) == []
+    end
+  end
+
+  describe "perform/1" do
+    setup do
+      # Stub IEM client so inline Oban doesn't blow up when executing enqueued jobs
+      Req.Test.stub(IemClient, fn conn ->
+        case conn.request_path do
+          "/cgi-bin/request/asos.py" -> Req.Test.text(conn, "#DEBUG,\nstation,valid,tmpf\n")
+          _ -> Req.Test.json(conn, %{"profiles" => []})
+        end
+      end)
+
+      :ok
+    end
+
+    test "enqueues weather jobs and marks QSOs as queued" do
+      _station = create_asos_station("KDFW", 32.90, -97.04)
+      qso = create_qso()
+
+      assert qso.weather_queued == false
+
+      assert :ok = QsoWeatherEnqueueWorker.perform(%Oban.Job{args: %{}})
+
+      updated = Repo.get!(Qso, qso.id)
+      assert updated.weather_queued == true
+    end
+
+    test "returns :ok when no unprocessed QSOs exist" do
+      assert :ok = QsoWeatherEnqueueWorker.perform(%Oban.Job{args: %{}})
+    end
+
+    test "skips QSOs without positions" do
+      _station = create_asos_station("KDFW", 32.90, -97.04)
+      _qso = create_qso(%{pos1: nil})
+
+      assert :ok = QsoWeatherEnqueueWorker.perform(%Oban.Job{args: %{}})
+
+      # QSO without pos1 should not be marked as queued
+      all_qsos = Repo.all(Qso)
+      refute Enum.any?(all_qsos, & &1.weather_queued)
+    end
+  end
+end