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)