From 40d1fa03aa05979b15083fb6dfe5b14511e17d8f Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Tue, 31 Mar 2026 12:29:39 -0500 Subject: [PATCH] QSO submission triggers enrichment directly, fix prod badarith crash - Add enqueue_for_qso/1 to directly enqueue weather/HRRR/terrain/IEMRE jobs for a single user-submitted QSO (no cron, no bulk processing) - Submit flow calls enqueue_for_qso instead of generic enqueue worker - Add enrichment queues to prod config for on-demand processing - Guard against HRRR fill values in store_hrrr_profiles (fixes badarith) - Filter QSOs without pos2 in build_terrain_jobs --- config/dev.exs | 2 +- config/runtime.exs | 10 +-- config/test.exs | 2 +- .../propagation/freshness_monitor.ex | 4 +- .../workers/propagation_grid_worker.ex | 4 +- .../workers/qso_weather_enqueue_worker.ex | 26 ++++++++ lib/microwaveprop_web/live/submit_live.ex | 10 +-- .../qso_weather_enqueue_worker_test.exs | 61 +++++++++++++++++++ 8 files changed, 100 insertions(+), 19 deletions(-) diff --git a/config/dev.exs b/config/dev.exs index 93533b11..90033455 100644 --- a/config/dev.exs +++ b/config/dev.exs @@ -11,7 +11,7 @@ config :microwaveprop, Microwaveprop.Repo, database: "microwaveprop_dev", stacktrace: true, show_sensitive_data_on_connection_error: true, - pool_size: 10 + pool_size: 30 # For development, we disable any cache and enable # debugging and code reloading. diff --git a/config/runtime.exs b/config/runtime.exs index 8063e8b1..7ceef372 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -65,12 +65,9 @@ if config_env() == :prod do ], secret_key_base: secret_key_base - config :microwaveprop, :dns_cluster_query, System.get_env("DNS_CLUSTER_QUERY") - config :microwaveprop, srtm_tiles_dir: "/srtm" - - # Production Oban: only live scoring and polling, no historical backfill + # Production Oban: live scoring, polling, and on-demand QSO enrichment (no cron backfill) config :microwaveprop, Oban, - queues: [propagation: 1, commercial: 2, solar: 1], + queues: [propagation: 1, commercial: 2, solar: 1, weather: 5, hrrr: 5, terrain: 2, iemre: 5], plugins: [ {Oban.Plugins.Pruner, max_age: 3600 * 24}, {Oban.Plugins.Lifeline, rescue_after: 30 * 60 * 1000}, @@ -83,6 +80,9 @@ if config_env() == :prod do ]} ] + config :microwaveprop, :dns_cluster_query, System.get_env("DNS_CLUSTER_QUERY") + config :microwaveprop, srtm_tiles_dir: "/srtm" + # ## SSL Support # # To get SSL working, you will need to add the `https` key diff --git a/config/test.exs b/config/test.exs index cdc7a1ac..9c6adb7f 100644 --- a/config/test.exs +++ b/config/test.exs @@ -28,7 +28,6 @@ config :microwaveprop, MicrowavepropWeb.Endpoint, # Run Oban jobs inline during tests config :microwaveprop, Oban, testing: :inline -config :microwaveprop, start_freshness_monitor: false config :microwaveprop, elevation_req_options: [plug: {Req.Test, Microwaveprop.Terrain.ElevationClient}, retry: false] config :microwaveprop, hrrr_req_options: [plug: {Req.Test, Microwaveprop.Weather.HrrrClient}, retry: false] config :microwaveprop, iem_req_options: [plug: {Req.Test, Microwaveprop.Weather.IemClient}, retry: false] @@ -36,6 +35,7 @@ config :microwaveprop, iem_req_options: [plug: {Req.Test, Microwaveprop.Weather. # Route HTTP requests through Req.Test stubs config :microwaveprop, solar_req_options: [plug: {Req.Test, Microwaveprop.Weather.SolarClient}] config :microwaveprop, srtm_req_options: [plug: {Req.Test, Microwaveprop.Terrain.Srtm}, retry: false] +config :microwaveprop, start_freshness_monitor: false # Initialize plugs at runtime for faster test compilation config :phoenix, :plug_init_mode, :runtime diff --git a/lib/microwaveprop/propagation/freshness_monitor.ex b/lib/microwaveprop/propagation/freshness_monitor.ex index 1b2647e6..9aea7485 100644 --- a/lib/microwaveprop/propagation/freshness_monitor.ex +++ b/lib/microwaveprop/propagation/freshness_monitor.ex @@ -12,7 +12,7 @@ defmodule Microwaveprop.Propagation.FreshnessMonitor do require Logger - @check_interval :timer.minutes(5) + @check_interval to_timeout(minute: 5) @stale_threshold_minutes 120 def start_link(_opts) do @@ -65,7 +65,7 @@ defmodule Microwaveprop.Propagation.FreshnessMonitor do ) ) - unless pending do + if !pending do Oban.insert(PropagationGridWorker.new(%{})) end end diff --git a/lib/microwaveprop/workers/propagation_grid_worker.ex b/lib/microwaveprop/workers/propagation_grid_worker.ex index 3ce66f55..64aa0d5e 100644 --- a/lib/microwaveprop/workers/propagation_grid_worker.ex +++ b/lib/microwaveprop/workers/propagation_grid_worker.ex @@ -59,7 +59,9 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do defp store_hrrr_profiles(grid_data, valid_time) do stored = Enum.flat_map(grid_data, fn {{lat, lon}, profile} -> - if profile.surface_temp_c do + temp = profile.surface_temp_c + + if temp && temp > -80 && temp < 60 do params = if is_list(profile.profile) and length(profile.profile) >= 3, do: SoundingParams.derive(profile.profile) diff --git a/lib/microwaveprop/workers/qso_weather_enqueue_worker.ex b/lib/microwaveprop/workers/qso_weather_enqueue_worker.ex index 720789ee..09666c96 100644 --- a/lib/microwaveprop/workers/qso_weather_enqueue_worker.ex +++ b/lib/microwaveprop/workers/qso_weather_enqueue_worker.ex @@ -15,6 +15,31 @@ defmodule Microwaveprop.Workers.QsoWeatherEnqueueWorker do # Oban jobs have ~149 params each; PG limit is 65535 params per query @insert_batch_size 400 + @doc """ + Enqueue all enrichment jobs (weather, HRRR, terrain, IEMRE) for a single QSO. + Called directly from submission flow — no Oban indirection. + """ + def enqueue_for_qso(qso) do + weather_jobs = build_weather_jobs([qso]) + hrrr_jobs = build_hrrr_jobs([qso]) + terrain_jobs = build_terrain_jobs([qso]) + iemre_jobs = build_iemre_jobs([qso]) + + all_jobs = weather_jobs ++ hrrr_jobs ++ terrain_jobs ++ iemre_jobs + + if all_jobs != [] do + insert_all_chunked(all_jobs) + end + + qso_ids = [qso.id] + if qso.pos1, do: Radio.mark_weather_queued!(qso_ids) + if qso.pos1, do: Radio.mark_hrrr_queued!(qso_ids) + if qso.pos1 && qso.pos2, do: Radio.mark_terrain_queued!(qso_ids) + if qso.pos1, do: Radio.mark_iemre_queued!(qso_ids) + + :ok + end + @impl Oban.Worker def perform(%Oban.Job{}) do enqueue_weather_jobs() @@ -87,6 +112,7 @@ defmodule Microwaveprop.Workers.QsoWeatherEnqueueWorker do def build_terrain_jobs(qsos) do qsos + |> Enum.filter(fn qso -> qso.pos1 && qso.pos2 end) |> Enum.map(fn qso -> TerrainProfileWorker.new(%{"qso_id" => qso.id}) end) diff --git a/lib/microwaveprop_web/live/submit_live.ex b/lib/microwaveprop_web/live/submit_live.ex index da47cd77..018ddca4 100644 --- a/lib/microwaveprop_web/live/submit_live.ex +++ b/lib/microwaveprop_web/live/submit_live.ex @@ -6,8 +6,6 @@ defmodule MicrowavepropWeb.SubmitLive do alias Microwaveprop.Radio.Qso alias Microwaveprop.Workers.QsoWeatherEnqueueWorker - require Logger - @band_options [ {"1296 MHz", "1296"}, {"2304 MHz", "2304"}, @@ -55,13 +53,7 @@ defmodule MicrowavepropWeb.SubmitLive do else case Radio.create_qso(qso_params) do {:ok, qso} -> - case Oban.insert(QsoWeatherEnqueueWorker.new(%{})) do - {:ok, _job} -> - :ok - - {:error, reason} -> - Logger.warning("Failed to enqueue weather processing: #{inspect(reason)}") - end + QsoWeatherEnqueueWorker.enqueue_for_qso(qso) {:noreply, socket diff --git a/test/microwaveprop/workers/qso_weather_enqueue_worker_test.exs b/test/microwaveprop/workers/qso_weather_enqueue_worker_test.exs index 6b28586c..d11969d1 100644 --- a/test/microwaveprop/workers/qso_weather_enqueue_worker_test.exs +++ b/test/microwaveprop/workers/qso_weather_enqueue_worker_test.exs @@ -560,4 +560,65 @@ defmodule Microwaveprop.Workers.QsoWeatherEnqueueWorkerTest do assert updated.iemre_queued == true end end + + describe "enqueue_for_qso/1" do + setup do + 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") + + "/iemre/" <> _ -> + Req.Test.json(conn, %{"data" => [%{"utc_hour" => 0, "p01m_mm" => 0.0}]}) + + _ -> + Req.Test.json(conn, %{"profiles" => []}) + end + end) + + Req.Test.stub(HrrrClient, fn conn -> + Plug.Conn.send_resp(conn, 404, "not found") + end) + + Req.Test.stub(ElevationClient, fn conn -> + params = Plug.Conn.fetch_query_params(conn).query_params + lat_count = params["latitude"] |> String.split(",") |> length() + Req.Test.json(conn, %{"elevation" => List.duplicate(200.0, lat_count)}) + end) + + :ok + end + + test "enqueues all enrichment jobs and marks QSO as fully queued" do + _station = create_asos_station("KDFW", 32.90, -97.04) + qso = create_qso() + + assert qso.weather_queued == false + assert qso.hrrr_queued == false + assert qso.terrain_queued == false + assert qso.iemre_queued == false + + assert :ok = QsoWeatherEnqueueWorker.enqueue_for_qso(qso) + + updated = Repo.get!(Qso, qso.id) + assert updated.weather_queued == true + assert updated.hrrr_queued == true + assert updated.terrain_queued == true + assert updated.iemre_queued == true + end + + test "works for QSO without pos2 (marks terrain_queued false)" do + _station = create_asos_station("KDFW", 32.90, -97.04) + qso = create_qso(%{pos2: nil}) + + assert :ok = QsoWeatherEnqueueWorker.enqueue_for_qso(qso) + + updated = Repo.get!(Qso, qso.id) + assert updated.weather_queued == true + assert updated.hrrr_queued == true + # terrain requires pos2, so it can't be queued + assert updated.terrain_queued == false + assert updated.iemre_queued == true + end + end end