From 525d9c4ea58017b48e567c122e7ecfb63b4ae166 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Fri, 24 Apr 2026 13:25:10 -0500 Subject: [PATCH] fix(weather): reconcile contact weather_status from SQL, not on view MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 2,019 contacts sat permanently in weather_status=:queued even after their ASOS data had been ingested. The only code path that flipped :queued → :complete was MicrowavepropWeb.ContactLive.Show on page view — contacts that nobody visited stayed :queued forever. Adds Weather.reconcile_weather_statuses/0, a single SQL UPDATE that flips every :queued contact whose ±2h / 150km window now contains at least one surface observation. Called at the tail of the existing ContactWeatherEnqueueWorker cron so the correction runs regardless of whether anyone opens the UI. Visible effect: the status page's "Weather done" count (previously capped at 79,975 while data kept landing) will converge on the real total after the next cron tick. 2838 tests + credo green. --- lib/microwaveprop/weather.ex | 58 +++++++++++++ .../workers/contact_weather_enqueue_worker.ex | 14 +++ test/microwaveprop/weather_test.exs | 87 +++++++++++++++++++ 3 files changed, 159 insertions(+) diff --git a/lib/microwaveprop/weather.ex b/lib/microwaveprop/weather.ex index e00ea761..d9fb115b 100644 --- a/lib/microwaveprop/weather.ex +++ b/lib/microwaveprop/weather.ex @@ -260,6 +260,64 @@ defmodule Microwaveprop.Weather do |> Repo.exists?() end + @doc """ + Flip `contacts.weather_status` from `:queued` to `:complete` for + every contact whose ±2h / 150km window now contains at least one + surface observation. Returns `{:ok, n}` where `n` is the number of + contacts advanced. + + Background: `WeatherFetchWorker` upserts observations but has no + back-pointer to the contacts that triggered the fetch — the same + obs row satisfies many contacts. Previously the only place that + flipped `:queued → :complete` was `MicrowavepropWeb.ContactLive.Show` + on page view, which left thousands of contacts stuck in `:queued` + even after their data had landed. This reconciler closes that loop + as a single SQL UPDATE, invoked from the hourly enqueuer cron. + + Radius encoded as a conservative ±1.5° lat/lon bounding box — the + same rectangle `weather_for_contact/2` uses at `radius_km: 150` + around the equator/mid-latitudes. + """ + @spec reconcile_weather_statuses() :: {:ok, non_neg_integer()} + def reconcile_weather_statuses do + sql = """ + UPDATE contacts c + SET weather_status = 'complete' + WHERE c.weather_status = 'queued' + AND c.pos1 IS NOT NULL + AND ( + EXISTS ( + SELECT 1 + FROM surface_observations o + JOIN weather_stations s ON s.id = o.station_id + WHERE o.observed_at >= c.qso_timestamp - interval '2 hours' + AND o.observed_at <= c.qso_timestamp + interval '2 hours' + AND s.lat BETWEEN ((c.pos1->>'lat')::float - 1.5) + AND ((c.pos1->>'lat')::float + 1.5) + AND s.lon BETWEEN ((c.pos1->>'lon')::float - 1.5) + AND ((c.pos1->>'lon')::float + 1.5) + ) + OR ( + c.pos2 IS NOT NULL + AND EXISTS ( + SELECT 1 + FROM surface_observations o + JOIN weather_stations s ON s.id = o.station_id + WHERE o.observed_at >= c.qso_timestamp - interval '2 hours' + AND o.observed_at <= c.qso_timestamp + interval '2 hours' + AND s.lat BETWEEN ((c.pos2->>'lat')::float - 1.5) + AND ((c.pos2->>'lat')::float + 1.5) + AND s.lon BETWEEN ((c.pos2->>'lon')::float - 1.5) + AND ((c.pos2->>'lon')::float + 1.5) + ) + ) + ) + """ + + %{num_rows: n} = Repo.query!(sql) + {:ok, n} + end + @doc """ True if the station already has at least one surface observation anywhere within the given UTC date. Used by the `asos_day` worker diff --git a/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex b/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex index dec54d3d..70c7b6b3 100644 --- a/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex +++ b/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex @@ -191,6 +191,20 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do enqueue_terrain_jobs() enqueue_iemre_jobs() + # After enqueuing, sweep any :queued contacts whose data has + # already landed (from a prior run's ingestion) but whose + # weather_status was never flipped. Without this, contacts stayed + # visible to the backfill scan forever. + case Weather.reconcile_weather_statuses() do + {:ok, n} when n > 0 -> + require Logger + + Logger.info("ContactWeatherEnqueueWorker: reconciled #{n} weather_status → :complete") + + _ -> + :ok + end + :ok end diff --git a/test/microwaveprop/weather_test.exs b/test/microwaveprop/weather_test.exs index a0e5d952..8d2ee684 100644 --- a/test/microwaveprop/weather_test.exs +++ b/test/microwaveprop/weather_test.exs @@ -26,6 +26,93 @@ defmodule Microwaveprop.WeatherTest do lon: -97.30 } + describe "reconcile_weather_statuses/0" do + alias Microwaveprop.Radio + alias Microwaveprop.Radio.Contact + + @contact_attrs %{ + station1: "W5XD", + station2: "K5TR", + qso_timestamp: ~U[2023-09-17 17:30:00Z], + mode: "CW", + band: Decimal.new("10000"), + 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_contact(attrs \\ %{}) do + {:ok, c} = + %Contact{} + |> Contact.changeset(Map.merge(@contact_attrs, attrs)) + |> Repo.insert() + + c + end + + test "flips queued → complete for contacts with obs in ±2h / 150km window" do + {:ok, station} = Weather.find_or_create_station(@station_attrs) + + Weather.upsert_surface_observation(station, %{ + observed_at: ~U[2023-09-17 17:45:00Z], + temp_f: 80.0 + }) + + contact = create_contact() + Radio.set_enrichment_status!([contact.id], :weather_status, :queued) + + {:ok, n} = Weather.reconcile_weather_statuses() + assert n == 1 + + assert %{weather_status: :complete} = Repo.get(Contact, contact.id) + end + + test "leaves :queued contacts without nearby obs alone" do + {:ok, station} = + Weather.find_or_create_station(%{ + @station_attrs + | station_code: "KSEA", + lat: 47.45, + lon: -122.31 + }) + + Weather.upsert_surface_observation(station, %{ + observed_at: ~U[2023-09-17 17:45:00Z], + temp_f: 60.0 + }) + + # Contact is in TX; KSEA obs doesn't cover its pos1/pos2. + contact = create_contact() + Radio.set_enrichment_status!([contact.id], :weather_status, :queued) + + {:ok, n} = Weather.reconcile_weather_statuses() + assert n == 0 + assert %{weather_status: :queued} = Repo.get(Contact, contact.id) + end + + test "leaves :complete and :pending contacts alone" do + {:ok, station} = Weather.find_or_create_station(@station_attrs) + + Weather.upsert_surface_observation(station, %{ + observed_at: ~U[2023-09-17 17:45:00Z], + temp_f: 80.0 + }) + + c_done = create_contact(%{station1: "DONE"}) + Radio.set_enrichment_status!([c_done.id], :weather_status, :complete) + + c_pending = create_contact(%{station1: "PEND"}) + Radio.set_enrichment_status!([c_pending.id], :weather_status, :pending) + + {:ok, _} = Weather.reconcile_weather_statuses() + + assert %{weather_status: :complete} = Repo.get(Contact, c_done.id) + assert %{weather_status: :pending} = Repo.get(Contact, c_pending.id) + end + end + describe "find_or_create_station/1" do test "creates a new station" do assert {:ok, station} = Weather.find_or_create_station(@station_attrs)