From 16883591b4041c6a205bd012cd8921cfba7cf544 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Sat, 4 Apr 2026 19:19:18 -0500 Subject: [PATCH] Database performance fixes and async backfill enqueue - Fix score_pressure crash on nil pressure_mb (coastal HRRR points) - Set 10-min timeout on grid score upsert transaction (was :infinity) - Single DELETE for prune_old_scores instead of N queries in a loop - Remove dead load_hrrr_refractivity that loaded 95k rows into nil map - Pass selected_time to point_detail to skip latest_valid_time sub-query - Batch station existence checks (1 query per path point, not per station) - Batch solar index upserts via insert_all in chunks of 500 - Batch backfill_distances via single UPDATE FROM VALUES statement - Add is_grid_point boolean + partial index to hrrr_profiles (replaces non-sargable modular arithmetic filter on every weather map query) - Add partial index on contacts(qso_timestamp) WHERE pos1 IS NOT NULL - Move backfill enqueue to Oban worker so UI returns immediately --- config/dev.exs | 12 ++- config/runtime.exs | 2 +- lib/microwaveprop/propagation.ex | 34 +++----- lib/microwaveprop/propagation/scorer.ex | 4 +- lib/microwaveprop/radio.ex | 43 ++++++---- lib/microwaveprop/weather.ex | 85 +++++++++++++++++-- lib/microwaveprop/weather/hrrr_profile.ex | 3 +- .../workers/asos_adjustment_worker.ex | 29 +------ .../workers/backfill_enqueue_worker.ex | 45 ++++++++++ .../workers/contact_weather_enqueue_worker.ex | 16 ++-- .../workers/solar_index_worker.ex | 4 +- lib/microwaveprop_web/live/backfill_live.ex | 35 ++------ lib/microwaveprop_web/live/map_live.ex | 2 +- ...416_add_is_grid_point_to_hrrr_profiles.exs | 21 +++++ ...001517_add_contacts_pos1_partial_index.exs | 10 +++ test/microwaveprop/weather_grid_test.exs | 5 ++ 16 files changed, 235 insertions(+), 115 deletions(-) create mode 100644 lib/microwaveprop/workers/backfill_enqueue_worker.ex create mode 100644 priv/repo/migrations/20260405001416_add_is_grid_point_to_hrrr_profiles.exs create mode 100644 priv/repo/migrations/20260405001517_add_contacts_pos1_partial_index.exs diff --git a/config/dev.exs b/config/dev.exs index 41e67089..8ba0df86 100644 --- a/config/dev.exs +++ b/config/dev.exs @@ -71,7 +71,17 @@ config :microwaveprop, MicrowavepropWeb.Endpoint, ] config :microwaveprop, Oban, - queues: [propagation: 1, solar: 1, weather: 20, enqueue: 1, hrrr: 5, terrain: 4, commercial: 2, iemre: 10], + queues: [ + propagation: 1, + solar: 1, + weather: 20, + enqueue: 1, + hrrr: 5, + terrain: 4, + commercial: 2, + iemre: 10, + backfill_enqueue: 1 + ], plugins: [ {Oban.Plugins.Pruner, max_age: 3600 * 24}, {Oban.Plugins.Lifeline, rescue_after: to_timeout(minute: 30)}, diff --git a/config/runtime.exs b/config/runtime.exs index 4a3c4cd7..08ef519a 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -117,7 +117,7 @@ if config_env() == :prod do # Production Oban: live scoring, polling, and on-demand QSO enrichment (no cron backfill) config :microwaveprop, Oban, - queues: [propagation: 1, commercial: 2, solar: 1, weather: 10, hrrr: 10, terrain: 5, iemre: 10], + queues: [propagation: 1, commercial: 2, solar: 1, weather: 10, hrrr: 10, terrain: 5, iemre: 10, backfill_enqueue: 1], plugins: [ {Oban.Plugins.Pruner, max_age: 3600 * 24}, {Oban.Plugins.Lifeline, rescue_after: 30 * 60 * 1000}, diff --git a/lib/microwaveprop/propagation.ex b/lib/microwaveprop/propagation.ex index 13ae877c..8b7c862e 100644 --- a/lib/microwaveprop/propagation.ex +++ b/lib/microwaveprop/propagation.ex @@ -146,7 +146,7 @@ defmodule Microwaveprop.Propagation do acc + count end) end, - timeout: :infinity + timeout: 600_000 ) if Keyword.get(opts, :prune, true) do @@ -159,28 +159,14 @@ defmodule Microwaveprop.Propagation do result end - @doc "Remove scores with valid_times older than 1 hour before the earliest current forecast." + @doc "Remove scores with valid_times older than 2 hours." def prune_old_scores do cutoff = DateTime.add(DateTime.utc_now(), -2, :hour) - # Get distinct old valid_times and delete one at a time to avoid massive single DELETEs - old_times = - Repo.all( - from(gs in GridScore, - where: gs.valid_time < ^cutoff, - select: gs.valid_time, - distinct: gs.valid_time - ) - ) + {deleted, _} = Repo.delete_all(from(gs in GridScore, where: gs.valid_time < ^cutoff), timeout: 120_000) - total = - Enum.reduce(old_times, 0, fn vt, acc -> - {deleted, _} = Repo.delete_all(from(gs in GridScore, where: gs.valid_time == ^vt)) - acc + deleted - end) - - if total > 0 do - Logger.info("PropagationScores: pruned #{total} old scores across #{length(old_times)} valid_times") + if deleted > 0 do + Logger.info("PropagationScores: pruned #{deleted} old scores (before #{cutoff})") end end @@ -265,21 +251,23 @@ defmodule Microwaveprop.Propagation do end @doc "Get the full score and factors for a specific grid point, snapped to nearest grid." - def point_detail(band_mhz, lat, lon) do + def point_detail(band_mhz, lat, lon, valid_time \\ nil) do step = Grid.step() snapped_lat = Float.round(Float.round(lat / step) * step, 3) snapped_lon = Float.round(Float.round(lon / step) * step, 3) - case latest_valid_time(band_mhz) do + time = valid_time || latest_valid_time(band_mhz) + + case time do nil -> nil - latest_time -> + _ -> Repo.one( from(gs in GridScore, where: gs.band_mhz == ^band_mhz and - gs.valid_time == ^latest_time and + gs.valid_time == ^time and gs.lat == ^snapped_lat and gs.lon == ^snapped_lon, select: %{ diff --git a/lib/microwaveprop/propagation/scorer.ex b/lib/microwaveprop/propagation/scorer.ex index eea077cf..8a4dfabc 100644 --- a/lib/microwaveprop/propagation/scorer.ex +++ b/lib/microwaveprop/propagation/scorer.ex @@ -303,7 +303,9 @@ defmodule Microwaveprop.Propagation.Scorer do Without previous reading, scores based on absolute pressure. With previous reading, scores based on pressure change (delta). """ - @spec score_pressure(number(), number() | nil) :: integer() + @spec score_pressure(number() | nil, number() | nil) :: integer() + def score_pressure(nil, _previous_mb), do: 50 + def score_pressure(current_mb, nil) do cond do current_mb < 1005 -> 80 diff --git a/lib/microwaveprop/radio.ex b/lib/microwaveprop/radio.ex index dcceb1fd..20a6e048 100644 --- a/lib/microwaveprop/radio.ex +++ b/lib/microwaveprop/radio.ex @@ -195,22 +195,35 @@ defmodule Microwaveprop.Radio do end def backfill_distances(contacts) do - Enum.each(contacts, fn contact -> - if is_nil(contact.distance_km) && contact.pos1 && contact.pos2 do - lat1 = contact.pos1["lat"] - lon1 = contact.pos1["lon"] || contact.pos1["lng"] - lat2 = contact.pos2["lat"] - lon2 = contact.pos2["lon"] || contact.pos2["lng"] - - if lat1 && lon1 && lat2 && lon2 do - dist = lat1 |> haversine_km(lon1, lat2, lon2) |> round() |> Decimal.new() - - Contact - |> where([q], q.id == ^contact.id) - |> Repo.update_all(set: [distance_km: dist]) - end + updates = + for contact <- contacts, + is_nil(contact.distance_km) && contact.pos1 && contact.pos2, + lat1 = contact.pos1["lat"], + lon1 = contact.pos1["lon"] || contact.pos1["lng"], + lat2 = contact.pos2["lat"], + lon2 = contact.pos2["lon"] || contact.pos2["lng"], + lat1 && lon1 && lat2 && lon2 do + dist = lat1 |> haversine_km(lon1, lat2, lon2) |> round() + {contact.id, dist} end - end) + + if updates != [] do + # Build a single UPDATE ... FROM (VALUES ...) statement + # Ecto.UUID.dump!/1 converts string UUIDs to the 16-byte binary Postgrex expects + {values_sql, params} = + updates + |> Enum.with_index() + |> Enum.reduce({"", []}, fn {{id, dist}, i}, {sql, params} -> + frag = "($#{i * 2 + 1}::uuid, $#{i * 2 + 2}::numeric)" + sep = if sql == "", do: "", else: ", " + {sql <> sep <> frag, params ++ [Ecto.UUID.dump!(id), dist]} + end) + + Repo.query!( + "UPDATE contacts AS c SET distance_km = v.dist FROM (VALUES #{values_sql}) AS v(id, dist) WHERE c.id = v.id", + params + ) + end end defp deg_to_rad(deg), do: deg * :math.pi() / 180 diff --git a/lib/microwaveprop/weather.ex b/lib/microwaveprop/weather.ex index 693c632d..e25fd05f 100644 --- a/lib/microwaveprop/weather.ex +++ b/lib/microwaveprop/weather.ex @@ -143,12 +143,80 @@ defmodule Microwaveprop.Weather do |> Repo.exists?() end + @doc "Returns a MapSet of station_ids that have surface observations in the time window." + def station_ids_with_surface_observations(station_ids, start_dt, end_dt) do + SurfaceObservation + |> where([o], o.station_id in ^station_ids) + |> where([o], o.observed_at >= ^start_dt and o.observed_at <= ^end_dt) + |> select([o], o.station_id) + |> distinct(true) + |> Repo.all() + |> MapSet.new() + end + def has_sounding?(station_id, observed_at) do Sounding |> where([s], s.station_id == ^station_id and s.observed_at == ^observed_at) |> Repo.exists?() end + @doc "Returns a MapSet of {station_id, observed_at} tuples that have soundings." + def station_ids_with_soundings(station_ids, sounding_times) do + Sounding + |> where([s], s.station_id in ^station_ids and s.observed_at in ^sounding_times) + |> select([s], {s.station_id, s.observed_at}) + |> distinct(true) + |> Repo.all() + |> MapSet.new() + end + + @doc "Batch upsert solar indices using insert_all in chunks of 500." + def upsert_solar_indices_batch(records) do + now = DateTime.truncate(DateTime.utc_now(), :second) + + records + |> Enum.chunk_every(500) + |> Enum.reduce(0, fn chunk, acc -> + entries = + Enum.map(chunk, fn attrs -> + %{ + id: Ecto.UUID.generate(), + date: attrs[:date] || attrs.date, + sfi: attrs[:sfi], + sfi_adjusted: attrs[:sfi_adjusted], + sunspot_number: attrs[:sunspot_number], + ap_index: attrs[:ap_index], + kp_values: attrs[:kp_values], + inserted_at: now, + updated_at: now + } + end) + + {count, _} = + Repo.insert_all(SolarIndex, entries, + on_conflict: + from(s in SolarIndex, + update: [ + set: [ + sfi: fragment("EXCLUDED.sfi"), + sfi_adjusted: fragment("EXCLUDED.sfi_adjusted"), + sunspot_number: fragment("EXCLUDED.sunspot_number"), + ap_index: fragment("EXCLUDED.ap_index"), + kp_values: fragment("EXCLUDED.kp_values"), + updated_at: fragment("EXCLUDED.updated_at") + ] + ], + where: + s.sfi != fragment("EXCLUDED.sfi") or + s.ap_index != fragment("EXCLUDED.ap_index") + ), + conflict_target: [:date] + ) + + acc + count + end) + end + def get_solar_index(date) do Repo.get_by(SolarIndex, date: date) end @@ -268,9 +336,7 @@ defmodule Microwaveprop.Weather do def latest_grid_valid_time do Repo.one( from(h in HrrrProfile, - where: - fragment("(? * 1000)::int % 125 = 0", h.lat) and - fragment("(? * 1000)::int % 125 = 0", h.lon), + where: h.is_grid_point == true, select: max(h.valid_time) ) ) @@ -285,10 +351,9 @@ defmodule Microwaveprop.Weather do from(h in HrrrProfile, where: h.valid_time == ^latest_vt and + h.is_grid_point == true and h.lat >= ^bounds["south"] and h.lat <= ^bounds["north"] and - h.lon >= ^bounds["west"] and h.lon <= ^bounds["east"] and - fragment("(? * 1000)::int % 125 = 0", h.lat) and - fragment("(? * 1000)::int % 125 = 0", h.lon), + h.lon >= ^bounds["west"] and h.lon <= ^bounds["east"], select: %{ lat: h.lat, lon: h.lon, @@ -369,8 +434,13 @@ defmodule Microwaveprop.Weather do |> Enum.reduce({0, nil}, fn chunk, {total_count, _} -> entries = Enum.map(chunk, fn attrs -> + is_gp = + rem(round(attrs.lat * 1000), 125) == 0 and + rem(round(attrs.lon * 1000), 125) == 0 + Map.merge(attrs, %{ id: Ecto.UUID.generate(), + is_grid_point: is_gp, inserted_at: now, updated_at: now }) @@ -465,8 +535,7 @@ defmodule Microwaveprop.Weather do """ DELETE FROM hrrr_profiles WHERE valid_time >= $2 AND valid_time < $1 - AND (lat * 1000)::int % 125 = 0 - AND (lon * 1000)::int % 125 = 0 + AND is_grid_point = true """, [cutoff, floor], timeout: 300_000 diff --git a/lib/microwaveprop/weather/hrrr_profile.ex b/lib/microwaveprop/weather/hrrr_profile.ex index fe725649..cfb05b28 100644 --- a/lib/microwaveprop/weather/hrrr_profile.ex +++ b/lib/microwaveprop/weather/hrrr_profile.ex @@ -22,12 +22,13 @@ defmodule Microwaveprop.Weather.HrrrProfile do field :min_refractivity_gradient, :float field :ducting_detected, :boolean, default: false field :duct_characteristics, {:array, :map} + field :is_grid_point, :boolean, default: false timestamps(type: :utc_datetime) end @required_fields ~w(valid_time lat lon)a - @optional_fields ~w(run_time profile hpbl_m pwat_mm surface_temp_c surface_dewpoint_c surface_pressure_mb surface_refractivity min_refractivity_gradient ducting_detected duct_characteristics)a + @optional_fields ~w(run_time profile hpbl_m pwat_mm surface_temp_c surface_dewpoint_c surface_pressure_mb surface_refractivity min_refractivity_gradient ducting_detected duct_characteristics is_grid_point)a def changeset(hrrr_profile, attrs) do hrrr_profile diff --git a/lib/microwaveprop/workers/asos_adjustment_worker.ex b/lib/microwaveprop/workers/asos_adjustment_worker.ex index a9509c9e..b81b40ab 100644 --- a/lib/microwaveprop/workers/asos_adjustment_worker.ex +++ b/lib/microwaveprop/workers/asos_adjustment_worker.ex @@ -96,9 +96,6 @@ defmodule Microwaveprop.Workers.AsosAdjustmentWorker do } end) - # Get the last HRRR refractivity data for blending - hrrr_refractivity = load_hrrr_refractivity() - # For each grid point, find nearest station and score Grid.conus_points() |> Task.async_stream( @@ -106,7 +103,7 @@ defmodule Microwaveprop.Workers.AsosAdjustmentWorker do nearest = find_nearest_station(obs_list, lat, lon) if nearest do - score_point_from_obs(point, nearest, valid_time, hrrr_refractivity) + score_point_from_obs(point, nearest, valid_time) end end, max_concurrency: System.schedulers_online() * 2, @@ -133,16 +130,13 @@ defmodule Microwaveprop.Workers.AsosAdjustmentWorker do end end - defp score_point_from_obs({lat, lon}, obs, valid_time, hrrr_refractivity) do + defp score_point_from_obs({lat, lon}, obs, valid_time) do temp_c = Scorer.f_to_c(obs.temp_f) dewpoint_c = Scorer.f_to_c(obs.dewpoint_f) if is_nil(temp_c) or is_nil(dewpoint_c) do [] else - # Get HRRR refractivity for this grid point (from last hourly computation) - refractivity = Map.get(hrrr_refractivity, {lat, lon}) - sky_pct = sky_condition_to_pct(obs.sky_condition) rain_rate = precip_to_rate(obs.precip_1h_in) @@ -159,7 +153,7 @@ defmodule Microwaveprop.Workers.AsosAdjustmentWorker do pressure_mb: obs.sea_level_pressure_mb, prev_pressure_mb: nil, rain_rate_mmhr: rain_rate, - min_refractivity_gradient: refractivity, + min_refractivity_gradient: nil, bl_depth_m: nil } @@ -178,23 +172,6 @@ defmodule Microwaveprop.Workers.AsosAdjustmentWorker do end end - # Load the refractivity gradient values from the last HRRR grid computation - defp load_hrrr_refractivity do - case Propagation.latest_valid_time() do - nil -> - %{} - - _time -> - # Use the 10 GHz band (arbitrary — refractivity factor is the same for all bands) - 10_000 - |> Propagation.latest_scores() - |> Enum.reduce(%{}, fn %{lat: lat, lon: lon} = _score, acc -> - # We don't store refractivity separately, so use nil (neutral score) - Map.put(acc, {lat, lon}, nil) - end) - end - end - defp sky_condition_to_pct(nil), do: nil defp sky_condition_to_pct("CLR"), do: 0.0 defp sky_condition_to_pct("SKC"), do: 0.0 diff --git a/lib/microwaveprop/workers/backfill_enqueue_worker.ex b/lib/microwaveprop/workers/backfill_enqueue_worker.ex new file mode 100644 index 00000000..22c57276 --- /dev/null +++ b/lib/microwaveprop/workers/backfill_enqueue_worker.ex @@ -0,0 +1,45 @@ +defmodule Microwaveprop.Workers.BackfillEnqueueWorker do + @moduledoc """ + Runs backfill enrichment enqueue as an Oban job so the backfill dashboard + returns immediately instead of blocking on the enqueue loop. + """ + use Oban.Worker, queue: :backfill_enqueue, max_attempts: 1 + + import Ecto.Query + + alias Microwaveprop.Radio.Contact + alias Microwaveprop.Repo + alias Microwaveprop.Workers.ContactWeatherEnqueueWorker + + require Logger + + @enrichable [:pending, :failed] + + @impl Oban.Worker + def perform(%Oban.Job{args: %{"limit" => limit}}) do + contacts = + Repo.all( + from(c in Contact, + where: + (not is_nil(c.pos1) or not is_nil(c.grid1)) and + (c.hrrr_status in ^@enrichable or c.weather_status in ^@enrichable or + c.terrain_status in ^@enrichable or c.iemre_status in ^@enrichable), + order_by: [desc: c.qso_timestamp], + limit: ^limit + ) + ) + + count = length(contacts) + Logger.info("BackfillEnqueue: processing #{count} contacts") + + Enum.each(contacts, &ContactWeatherEnqueueWorker.enqueue_for_contact/1) + + Phoenix.PubSub.broadcast( + Microwaveprop.PubSub, + "backfill:enqueue_complete", + {:enqueue_complete, count} + ) + + :ok + end +end diff --git a/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex b/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex index 76e2c6e5..d8f18fe2 100644 --- a/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex +++ b/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex @@ -198,11 +198,12 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker 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.reject(fn station -> - Weather.has_surface_observations?(station.id, start_dt, end_dt) - end) + stations = Weather.nearby_stations(lat, lon, "asos", @asos_radius_km) + station_ids = Enum.map(stations, & &1.id) + covered = Weather.station_ids_with_surface_observations(station_ids, start_dt, end_dt) + + stations + |> Enum.reject(fn station -> MapSet.member?(covered, station.id) end) |> Enum.map(fn station -> WeatherFetchWorker.new(%{ "fetch_type" => "asos", @@ -240,12 +241,13 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do 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) + station_ids = Enum.map(stations, & &1.id) + covered = Weather.station_ids_with_soundings(station_ids, sounding_times) for station <- stations, sounding_time <- sounding_times, - not Weather.has_sounding?(station.id, sounding_time) do + not MapSet.member?(covered, {station.id, sounding_time}) do WeatherFetchWorker.new(%{ "fetch_type" => "raob", "station_id" => station.id, diff --git a/lib/microwaveprop/workers/solar_index_worker.ex b/lib/microwaveprop/workers/solar_index_worker.ex index 9cde7924..25a619b7 100644 --- a/lib/microwaveprop/workers/solar_index_worker.ex +++ b/lib/microwaveprop/workers/solar_index_worker.ex @@ -15,7 +15,7 @@ defmodule Microwaveprop.Workers.SolarIndexWorker do case SolarClient.fetch_solar_indices() do {:ok, all_records} -> matched = Enum.filter(all_records, fn r -> r.date == target_date end) - Enum.each(matched, &Weather.upsert_solar_index/1) + Weather.upsert_solar_indices_batch(matched) if matched != [] do Phoenix.PubSub.broadcast( @@ -39,7 +39,7 @@ defmodule Microwaveprop.Workers.SolarIndexWorker do {:ok, all_records} -> all_records |> SolarClient.filter_since(since_date) - |> Enum.each(&Weather.upsert_solar_index/1) + |> Weather.upsert_solar_indices_batch() :ok diff --git a/lib/microwaveprop_web/live/backfill_live.ex b/lib/microwaveprop_web/live/backfill_live.ex index 237143a0..c9229fa1 100644 --- a/lib/microwaveprop_web/live/backfill_live.ex +++ b/lib/microwaveprop_web/live/backfill_live.ex @@ -7,13 +7,14 @@ defmodule MicrowavepropWeb.BackfillLive do alias Microwaveprop.Radio.Contact alias Microwaveprop.Repo - alias Microwaveprop.Workers.ContactWeatherEnqueueWorker + alias Microwaveprop.Workers.BackfillEnqueueWorker @impl true def mount(_params, _session, socket) do if connected?(socket) do Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "db:contact_status") Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "db:oban_jobs") + Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "backfill:enqueue_complete") end stats = fetch_stats() @@ -43,19 +44,15 @@ defmodule MicrowavepropWeb.BackfillLive do def handle_event("enqueue", %{"limit" => limit_str}, socket) do limit = String.to_integer(limit_str) - # Run in a task to not block the LiveView - pid = self() - - Task.start(fn -> - count = enqueue_batch(limit) - send(pid, {:enqueued, count}) - end) + %{"limit" => limit} + |> BackfillEnqueueWorker.new() + |> Oban.insert() {:noreply, socket |> assign(enqueuing: true, limit: limit) |> LiveStash.stash_assigns([:limit])} end @impl true - def handle_info({:enqueued, count}, socket) do + def handle_info({:enqueue_complete, count}, socket) do {:noreply, assign(socket, last_enqueued: count, enqueuing: false)} end @@ -87,26 +84,6 @@ defmodule MicrowavepropWeb.BackfillLive do assign(socket, refresh_timer: ref) end - @enrichable [:pending, :failed] - - defp enqueue_batch(limit) do - contacts = - Repo.all( - from(c in Contact, - where: - (not is_nil(c.pos1) or not is_nil(c.grid1)) and - (c.hrrr_status in ^@enrichable or c.weather_status in ^@enrichable or c.terrain_status in ^@enrichable or - c.iemre_status in ^@enrichable), - order_by: [desc: c.qso_timestamp], - limit: ^limit - ) - ) - - # ensure_positions! is called inside enqueue_for_contact - Enum.each(contacts, &ContactWeatherEnqueueWorker.enqueue_for_contact/1) - length(contacts) - end - defp count_unprocessed do incomplete = [:pending, :queued, :processing, :failed] diff --git a/lib/microwaveprop_web/live/map_live.ex b/lib/microwaveprop_web/live/map_live.ex index 918ec3ca..944d390c 100644 --- a/lib/microwaveprop_web/live/map_live.ex +++ b/lib/microwaveprop_web/live/map_live.ex @@ -106,7 +106,7 @@ defmodule MicrowavepropWeb.MapLive do def handle_event("point_detail", %{"lat" => lat, "lon" => lon}, socket) do band = socket.assigns.selected_band - detail = Propagation.point_detail(band, lat, lon) + detail = Propagation.point_detail(band, lat, lon, socket.assigns.selected_time) forecast = Propagation.point_forecast(band, lat, lon) forecast_data = diff --git a/priv/repo/migrations/20260405001416_add_is_grid_point_to_hrrr_profiles.exs b/priv/repo/migrations/20260405001416_add_is_grid_point_to_hrrr_profiles.exs new file mode 100644 index 00000000..8bc61d42 --- /dev/null +++ b/priv/repo/migrations/20260405001416_add_is_grid_point_to_hrrr_profiles.exs @@ -0,0 +1,21 @@ +defmodule Microwaveprop.Repo.Migrations.AddIsGridPointToHrrrProfiles do + use Ecto.Migration + + def change do + alter table(:hrrr_profiles) do + add :is_grid_point, :boolean, default: false, null: false + end + + flush() + + execute( + "UPDATE hrrr_profiles SET is_grid_point = true WHERE (lat * 1000)::int % 125 = 0 AND (lon * 1000)::int % 125 = 0", + "SELECT 1" + ) + + create index(:hrrr_profiles, [:valid_time], + where: "is_grid_point = true", + name: :hrrr_profiles_grid_point_valid_time_idx + ) + end +end diff --git a/priv/repo/migrations/20260405001517_add_contacts_pos1_partial_index.exs b/priv/repo/migrations/20260405001517_add_contacts_pos1_partial_index.exs new file mode 100644 index 00000000..cc3b601e --- /dev/null +++ b/priv/repo/migrations/20260405001517_add_contacts_pos1_partial_index.exs @@ -0,0 +1,10 @@ +defmodule Microwaveprop.Repo.Migrations.AddContactsPos1PartialIndex do + use Ecto.Migration + + def change do + create index(:contacts, [:qso_timestamp], + where: "pos1 IS NOT NULL", + name: :contacts_pos1_not_null_idx + ) + end +end diff --git a/test/microwaveprop/weather_grid_test.exs b/test/microwaveprop/weather_grid_test.exs index 8a383c7f..c2ae1e80 100644 --- a/test/microwaveprop/weather_grid_test.exs +++ b/test/microwaveprop/weather_grid_test.exs @@ -245,6 +245,10 @@ defmodule Microwaveprop.WeatherGridTest do defp insert_hrrr_profile(attrs) do now = DateTime.truncate(DateTime.utc_now(), :second) + is_gp = + rem(round(attrs.lat * 1000), 125) == 0 and + rem(round(attrs.lon * 1000), 125) == 0 + defaults = %{ id: Ecto.UUID.bingenerate(), run_time: DateTime.add(attrs.valid_time, -3600, :second), @@ -253,6 +257,7 @@ defmodule Microwaveprop.WeatherGridTest do min_refractivity_gradient: nil, ducting_detected: nil, duct_characteristics: nil, + is_grid_point: is_gp, inserted_at: now, updated_at: now }