diff --git a/lib/microwaveprop/application.ex b/lib/microwaveprop/application.ex index 4e267ecd..c5f27a37 100644 --- a/lib/microwaveprop/application.ex +++ b/lib/microwaveprop/application.ex @@ -16,6 +16,7 @@ defmodule Microwaveprop.Application do {Phoenix.PubSub, name: Microwaveprop.PubSub}, Microwaveprop.Cache, Microwaveprop.Propagation.ScoreCache, + Microwaveprop.Weather.GridCache, Microwaveprop.Weather.NexradCache, {Oban, Application.fetch_env!(:microwaveprop, Oban)}, Microwaveprop.RepoListener, diff --git a/lib/microwaveprop/weather.ex b/lib/microwaveprop/weather.ex index 5b7a6c6d..a9813bc8 100644 --- a/lib/microwaveprop/weather.ex +++ b/lib/microwaveprop/weather.ex @@ -5,6 +5,7 @@ defmodule Microwaveprop.Weather do alias Microwaveprop.Repo alias Microwaveprop.Weather.Era5Profile + alias Microwaveprop.Weather.GridCache alias Microwaveprop.Weather.HrrrProfile alias Microwaveprop.Weather.IemClient alias Microwaveprop.Weather.IemreObservation @@ -359,12 +360,18 @@ defmodule Microwaveprop.Weather do @spec latest_grid_valid_time() :: DateTime.t() | nil def latest_grid_valid_time do - Repo.one( - from(h in HrrrProfile, - where: h.is_grid_point == true, - select: max(h.valid_time) - ) - ) + case GridCache.latest_valid_time() do + %DateTime{} = vt -> + vt + + nil -> + Repo.one( + from(h in HrrrProfile, + where: h.is_grid_point == true, + select: max(h.valid_time) + ) + ) + end end @spec latest_weather_grid(map()) :: [map()] @@ -374,41 +381,77 @@ defmodule Microwaveprop.Weather do [] latest_vt -> - 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"], - select: %{ - lat: h.lat, - lon: h.lon, - valid_time: h.valid_time, - temperature: h.surface_temp_c, - dewpoint_depression: fragment("? - ?", h.surface_temp_c, h.surface_dewpoint_c), - bl_height: h.hpbl_m, - pwat: h.pwat_mm, - refractivity_gradient: h.min_refractivity_gradient, - ducting: h.ducting_detected, - surface_pressure_mb: h.surface_pressure_mb, - surface_temp_c: h.surface_temp_c, - surface_dewpoint_c: h.surface_dewpoint_c, - surface_refractivity: h.surface_refractivity, - profile: h.profile, - duct_characteristics: h.duct_characteristics - } - ) - |> Repo.all() - |> Enum.map(&derive_and_clean/1) + case GridCache.fetch_bounds(latest_vt, bounds) do + {:ok, rows} -> + rows + + :miss -> + full = load_weather_grid_from_db(latest_vt) + GridCache.put(latest_vt, full) + filter_weather_bounds(full, bounds) + end end end + defp load_weather_grid_from_db(latest_vt) do + from(h in HrrrProfile, + where: h.valid_time == ^latest_vt and h.is_grid_point == true, + select: %{ + lat: h.lat, + lon: h.lon, + valid_time: h.valid_time, + temperature: h.surface_temp_c, + dewpoint_depression: fragment("? - ?", h.surface_temp_c, h.surface_dewpoint_c), + bl_height: h.hpbl_m, + pwat: h.pwat_mm, + refractivity_gradient: h.min_refractivity_gradient, + ducting: h.ducting_detected, + surface_pressure_mb: h.surface_pressure_mb, + surface_temp_c: h.surface_temp_c, + surface_dewpoint_c: h.surface_dewpoint_c, + surface_refractivity: h.surface_refractivity, + profile: h.profile, + duct_characteristics: h.duct_characteristics + } + ) + |> Repo.all() + |> Enum.map(&derive_and_clean/1) + end + + defp filter_weather_bounds(rows, nil), do: rows + + defp filter_weather_bounds(rows, %{"south" => s, "north" => n, "west" => w, "east" => e}) do + Enum.filter(rows, fn %{lat: lat, lon: lon} -> + lat >= s and lat <= n and lon >= w and lon <= e + end) + end + + @doc """ + Eagerly populate the `GridCache` with the full CONUS weather grid for + `valid_time` and broadcast it to every node in the cluster. Called from + `PropagationGridWorker` after each hourly HRRR upsert so connected clients + see zero-DB pan/zoom latency on the `/weather` map. + """ + @spec warm_grid_cache_and_broadcast(DateTime.t()) :: :ok + def warm_grid_cache_and_broadcast(valid_time) do + rows = load_weather_grid_from_db(valid_time) + GridCache.broadcast_put(valid_time, rows) + :ok + end + @spec weather_point_detail(float(), float(), DateTime.t()) :: map() | nil def weather_point_detail(lat, lon, valid_time) do step = 0.125 snapped_lat = Float.round(Float.round(lat / step) * step, 3) snapped_lon = Float.round(Float.round(lon / step) * step, 3) + case GridCache.fetch_point(valid_time, snapped_lat, snapped_lon) do + {:ok, row} -> row + :miss -> weather_point_detail_from_db(valid_time, snapped_lat, snapped_lon) + end + end + + defp weather_point_detail_from_db(valid_time, snapped_lat, snapped_lon) do from(h in HrrrProfile, where: h.lat == ^snapped_lat and h.lon == ^snapped_lon and h.valid_time == ^valid_time, select: %{ diff --git a/lib/microwaveprop/weather/grid_cache.ex b/lib/microwaveprop/weather/grid_cache.ex new file mode 100644 index 00000000..16aaa6b5 --- /dev/null +++ b/lib/microwaveprop/weather/grid_cache.ex @@ -0,0 +1,137 @@ +defmodule Microwaveprop.Weather.GridCache do + @moduledoc """ + Node-local ETS cache of derived HRRR grid rows keyed by `valid_time`. Mirrors + `Microwaveprop.Propagation.ScoreCache` but for the `/weather` map. + + The Weather map LiveView calls `latest_weather_grid/1` on mount and every + pan/zoom. Each call otherwise hits the 42M-row partitioned `hrrr_profiles` + table, runs per-row `derive_and_clean` transforms, and returns 3-10k rows. + With this cache those calls become in-memory map iterations. + + Each cache entry stores `%{{lat, lon} => derived_row}` so per-point lookups + (used by `weather_point_detail/3`) are O(1). Populated by + `Microwaveprop.Weather.warm_grid_cache/1` after the hourly worker upserts + new HRRR data, fanned out across the cluster via the `"weather:cache"` + PubSub topic so every node stays in sync. + """ + use GenServer + + alias Phoenix.PubSub + + @table :weather_grid_cache + @topic "weather:cache" + @pubsub Microwaveprop.PubSub + + @type row :: %{required(:lat) => float(), required(:lon) => float(), optional(atom()) => any()} + @type bounds :: %{optional(String.t()) => float()} + + @spec start_link(keyword()) :: GenServer.on_start() + def start_link(opts), do: GenServer.start_link(__MODULE__, opts, name: __MODULE__) + + @spec fetch(DateTime.t()) :: {:ok, [row()]} | :miss + def fetch(valid_time) do + case :ets.lookup(@table, valid_time) do + [{_, grid}] -> {:ok, grid_to_list(grid)} + [] -> :miss + end + end + + @spec fetch_bounds(DateTime.t(), bounds() | nil) :: {:ok, [row()]} | :miss + def fetch_bounds(valid_time, bounds) do + case :ets.lookup(@table, valid_time) do + [{_, grid}] -> {:ok, grid_to_filtered_list(grid, bounds)} + [] -> :miss + end + end + + @spec fetch_point(DateTime.t(), float(), float()) :: {:ok, row()} | :miss + def fetch_point(valid_time, lat, lon) do + case :ets.lookup(@table, valid_time) do + [{_, grid}] -> + case Map.get(grid, {lat, lon}) do + nil -> :miss + row -> {:ok, row} + end + + [] -> + :miss + end + end + + @spec put(DateTime.t(), [row()]) :: :ok + def put(valid_time, rows) do + grid = list_to_grid(rows) + :ets.insert(@table, {valid_time, grid}) + :ok + end + + @doc "Insert locally AND broadcast to peer nodes via PubSub." + @spec broadcast_put(DateTime.t(), [row()]) :: :ok + def broadcast_put(valid_time, rows) do + PubSub.broadcast(@pubsub, @topic, {:weather_cache_refresh, valid_time, rows}) + :ok + end + + @spec latest_valid_time() :: DateTime.t() | nil + def latest_valid_time do + match_spec = [{{:"$1", :_}, [], [:"$1"]}] + + case :ets.select(@table, match_spec) do + [] -> nil + times -> Enum.max(times, DateTime) + end + end + + @spec prune_older_than(DateTime.t()) :: non_neg_integer() + def prune_older_than(cutoff) do + match_spec = [{{:"$1", :_}, [{:<, :"$1", {:const, cutoff}}], [true]}] + :ets.select_delete(@table, match_spec) + end + + @spec clear() :: :ok + def clear do + :ets.delete_all_objects(@table) + :ok + end + + @spec sync() :: :ok + def sync do + GenServer.call(__MODULE__, :sync) + end + + @impl true + def init(_opts) do + :ets.new(@table, [:set, :named_table, :public, read_concurrency: true]) + PubSub.subscribe(@pubsub, @topic) + {:ok, %{}} + end + + @impl true + def handle_call(:sync, _from, state), do: {:reply, :ok, state} + + @impl true + def handle_info({:weather_cache_refresh, valid_time, rows}, state) do + put(valid_time, rows) + {:noreply, state} + end + + def handle_info(_msg, state), do: {:noreply, state} + + # ---------- Internal ---------- + + defp list_to_grid(rows) do + Map.new(rows, fn %{lat: lat, lon: lon} = row -> {{lat, lon}, row} end) + end + + defp grid_to_list(grid), do: Enum.map(grid, fn {_, row} -> row end) + + defp grid_to_filtered_list(grid, nil), do: grid_to_list(grid) + + defp grid_to_filtered_list(grid, %{"south" => s, "north" => n, "west" => w, "east" => e}) do + grid + |> Enum.filter(fn {{lat, lon}, _} -> + lat >= s and lat <= n and lon >= w and lon <= e + end) + |> Enum.map(fn {_, row} -> row end) + end +end diff --git a/lib/microwaveprop/workers/propagation_grid_worker.ex b/lib/microwaveprop/workers/propagation_grid_worker.ex index b105fd6e..ba6fd19f 100644 --- a/lib/microwaveprop/workers/propagation_grid_worker.ex +++ b/lib/microwaveprop/workers/propagation_grid_worker.ex @@ -85,6 +85,8 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do grid_data = merge_native_duct_data(grid_data, run_time, forecast_hour) :erlang.garbage_collect() + Weather.warm_grid_cache_and_broadcast(valid_time) + Phoenix.PubSub.broadcast( Microwaveprop.PubSub, "weather:updated", diff --git a/lib/microwaveprop_web/live/backfill_live.ex b/lib/microwaveprop_web/live/backfill_live.ex index 09554e84..743c6b35 100644 --- a/lib/microwaveprop_web/live/backfill_live.ex +++ b/lib/microwaveprop_web/live/backfill_live.ex @@ -5,6 +5,7 @@ defmodule MicrowavepropWeb.BackfillLive do import Ecto.Query + alias Microwaveprop.Cache alias Microwaveprop.Radio.Contact alias Microwaveprop.Repo alias Microwaveprop.Workers.BackfillEnqueueWorker @@ -92,11 +93,19 @@ defmodule MicrowavepropWeb.BackfillLive do end defp schedule_refresh(socket) do - ref = Process.send_after(self(), :refresh_stats, 1000) + # 2s debounce — bulk enrichment runs produce hundreds of status_changed + # events per second, and each stats refresh fires 8+ count queries. + ref = Process.send_after(self(), :refresh_stats, 2_000) assign(socket, refresh_timer: ref) end defp count_unprocessed do + Cache.fetch_or_store({__MODULE__, :count_unprocessed}, 2_000, fn -> + count_unprocessed_raw() + end) + end + + defp count_unprocessed_raw do incomplete = [:pending, :queued, :processing, :failed] done = [:complete, :unavailable] @@ -147,6 +156,10 @@ defmodule MicrowavepropWeb.BackfillLive do end defp fetch_stats do + Cache.fetch_or_store({__MODULE__, :stats}, 2_000, fn -> fetch_stats_raw() end) + end + + defp fetch_stats_raw do jobs = Repo.all( from(j in "oban_jobs", @@ -190,7 +203,12 @@ defmodule MicrowavepropWeb.BackfillLive do end defp fetch_db_stats do - # Regular tables + partitioned tables (summing child partition stats) + # pg_class + pg_stat_user_tables join is slow — 5s cache is fine since + # table row counts barely change in a 5-second window. + Cache.fetch_or_store({__MODULE__, :db_stats}, 5_000, fn -> fetch_db_stats_raw() end) + end + + defp fetch_db_stats_raw do %{rows: rows} = Repo.query!(""" SELECT diff --git a/lib/microwaveprop_web/live/contact_live/show.ex b/lib/microwaveprop_web/live/contact_live/show.ex index eee57eec..6b880a80 100644 --- a/lib/microwaveprop_web/live/contact_live/show.ex +++ b/lib/microwaveprop_web/live/contact_live/show.ex @@ -35,37 +35,29 @@ defmodule MicrowavepropWeb.ContactLive.Show do Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "contact_enrichment:hrrr") Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "contact_enrichment:weather") Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "contact_enrichment:solar") + + # Defer the expensive enrichment loads (HRRR profiles, terrain profile, + # IEMRE, elevation profile, propagation analysis — each touching large + # tables or doing ITU-R math) until after the initial shell has been + # sent. Users see the contact basics in ~20ms instead of waiting the + # full ~500ms-2s mount. + send(self(), {:hydrate, can_enqueue}) end - weather = load_weather(contact) - solar = if can_enqueue, do: maybe_enqueue_solar(contact), else: load_solar(contact) - hrrr_path = Weather.hrrr_profiles_for_path(contact) - hrrr = List.first(hrrr_path) - - {hrrr, contact} = - if can_enqueue, do: maybe_enqueue_hrrr(hrrr, contact), else: {hrrr, contact} - - terrain = Terrain.get_terrain_profile(contact.id) - contact = if can_enqueue, do: maybe_enqueue_terrain(terrain, contact), else: contact - contact = if can_enqueue, do: maybe_enqueue_weather(weather, contact), else: contact - iemre = load_iemre(contact) - elevation_profile = compute_elevation_profile(contact, hrrr_path, weather.soundings) - propagation_analysis = build_propagation_analysis(contact, hrrr_path, terrain, elevation_profile, weather.soundings) - data_sources = build_data_sources(hrrr, hrrr_path, terrain, weather, elevation_profile) - {:ok, assign(socket, page_title: "#{contact.station1} / #{contact.station2}", contact: contact, - surface_observations: weather.surface_observations, - soundings: weather.soundings, - solar: solar, - hrrr: hrrr, - iemre: iemre, - terrain: terrain, - elevation_profile: elevation_profile, - propagation_analysis: propagation_analysis, - data_sources: data_sources, + surface_observations: [], + soundings: [], + solar: nil, + hrrr: nil, + iemre: nil, + terrain: nil, + elevation_profile: nil, + propagation_analysis: nil, + data_sources: nil, + hydrated: false, terrain_expanded: false, hrrr_profile_expanded: false, obs_sort_by: "station_name", @@ -234,6 +226,44 @@ defmodule MicrowavepropWeb.ContactLive.Show do end @impl true + def handle_info({:hydrate, can_enqueue}, socket) do + contact = socket.assigns.contact + + weather = load_weather(contact) + solar = if can_enqueue, do: maybe_enqueue_solar(contact), else: load_solar(contact) + hrrr_path = Weather.hrrr_profiles_for_path(contact) + hrrr = List.first(hrrr_path) + + {hrrr, contact} = + if can_enqueue, do: maybe_enqueue_hrrr(hrrr, contact), else: {hrrr, contact} + + terrain = Terrain.get_terrain_profile(contact.id) + contact = if can_enqueue, do: maybe_enqueue_terrain(terrain, contact), else: contact + contact = if can_enqueue, do: maybe_enqueue_weather(weather, contact), else: contact + iemre = load_iemre(contact) + elevation_profile = compute_elevation_profile(contact, hrrr_path, weather.soundings) + + propagation_analysis = + build_propagation_analysis(contact, hrrr_path, terrain, elevation_profile, weather.soundings) + + data_sources = build_data_sources(hrrr, hrrr_path, terrain, weather, elevation_profile) + + {:noreply, + assign(socket, + contact: contact, + surface_observations: weather.surface_observations, + soundings: weather.soundings, + solar: solar, + hrrr: hrrr, + iemre: iemre, + terrain: terrain, + elevation_profile: elevation_profile, + propagation_analysis: propagation_analysis, + data_sources: data_sources, + hydrated: true + )} + end + def handle_info({:terrain_ready, _contact_id}, socket) do contact = %{socket.assigns.contact | terrain_status: :complete} terrain = Terrain.get_terrain_profile(contact.id) @@ -502,15 +532,17 @@ defmodule MicrowavepropWeb.ContactLive.Show do end defp fetch_queue_counts do - import Ecto.Query + Microwaveprop.Cache.fetch_or_store({__MODULE__, :queue_counts}, 5_000, fn -> + import Ecto.Query - from(j in Oban.Job, - where: j.state in ["available", "scheduled", "retryable", "executing"], - group_by: j.queue, - select: {j.queue, count(j.id)} - ) - |> Microwaveprop.Repo.all() - |> Map.new() + from(j in Oban.Job, + where: j.state in ["available", "scheduled", "retryable", "executing"], + group_by: j.queue, + select: {j.queue, count(j.id)} + ) + |> Microwaveprop.Repo.all() + |> Map.new() + end) end defp internal_network?(session) do diff --git a/test/microwaveprop/weather/grid_cache_test.exs b/test/microwaveprop/weather/grid_cache_test.exs new file mode 100644 index 00000000..31e45fe4 --- /dev/null +++ b/test/microwaveprop/weather/grid_cache_test.exs @@ -0,0 +1,85 @@ +defmodule Microwaveprop.Weather.GridCacheTest do + use ExUnit.Case, async: false + + alias Microwaveprop.Weather.GridCache + + setup do + GridCache.clear() + :ok + end + + describe "fetch/1" do + test "returns :miss when nothing is cached" do + assert GridCache.fetch(~U[2026-04-12 12:00:00Z]) == :miss + end + + test "returns cached rows after put" do + rows = [%{lat: 32.0, lon: -97.0, temperature: 25.0}] + GridCache.put(~U[2026-04-12 12:00:00Z], rows) + assert {:ok, [%{lat: 32.0, lon: -97.0}]} = GridCache.fetch(~U[2026-04-12 12:00:00Z]) + end + end + + describe "fetch_bounds/2" do + setup do + rows = [ + %{lat: 32.0, lon: -97.0, temperature: 25.0}, + %{lat: 40.0, lon: -74.0, temperature: 20.0}, + %{lat: 34.0, lon: -98.0, temperature: 28.0} + ] + + GridCache.put(~U[2026-04-12 12:00:00Z], rows) + :ok + end + + test "returns all rows when bounds are nil" do + assert {:ok, list} = GridCache.fetch_bounds(~U[2026-04-12 12:00:00Z], nil) + assert length(list) == 3 + end + + test "returns only rows inside the bounds" do + bounds = %{"south" => 31.0, "north" => 35.0, "west" => -100.0, "east" => -95.0} + assert {:ok, list} = GridCache.fetch_bounds(~U[2026-04-12 12:00:00Z], bounds) + assert length(list) == 2 + end + + test "returns :miss when the valid_time is not cached" do + bounds = %{"south" => 0.0, "north" => 90.0, "west" => -180.0, "east" => 0.0} + assert GridCache.fetch_bounds(~U[2099-01-01 00:00:00Z], bounds) == :miss + end + end + + describe "fetch_point/3" do + setup do + rows = [ + %{lat: 32.0, lon: -97.0, temperature: 25.0}, + %{lat: 33.0, lon: -97.0, temperature: 27.0} + ] + + GridCache.put(~U[2026-04-12 12:00:00Z], rows) + :ok + end + + test "returns the cached row for a known point" do + assert {:ok, %{temperature: 25.0}} = + GridCache.fetch_point(~U[2026-04-12 12:00:00Z], 32.0, -97.0) + end + + test "returns :miss for an unknown point" do + assert GridCache.fetch_point(~U[2026-04-12 12:00:00Z], 40.0, -74.0) == :miss + end + end + + describe "latest_valid_time/0" do + test "returns the most recent cached valid_time" do + GridCache.put(~U[2026-04-12 10:00:00Z], []) + GridCache.put(~U[2026-04-12 14:00:00Z], []) + GridCache.put(~U[2026-04-12 12:00:00Z], []) + assert GridCache.latest_valid_time() == ~U[2026-04-12 14:00:00Z] + end + + test "returns nil when nothing is cached" do + assert GridCache.latest_valid_time() == nil + end + end +end